PyHealth

https://github.com/sunlabuiuc/pyhealth
Admin
PyHealth is a comprehensive Python toolkit for healthcare AI that enables building, testing, and...
Tokens:1,126,390
Snippets:8,957
Trust Score:8.2
License:MIT
Update:4 months ago
Show doc for...
Tokens:
Raw
### Environment Setup and Initialization

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/eeg/sleep_staging/contrawr_sleepedf.ipynb

Seeds random generators, imports core dependencies, and detects the training device (CPU or CUDA). Ensure PyTorch and NumPy are installed.

```python
import random

import numpy as np
import torch

from pyhealth.datasets import SleepEDFDataset
from pyhealth.datasets.splitter import split_by_sample
from pyhealth.datasets.utils import get_dataloader

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### TUEVDataset Initialization and Usage Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Provides a complete example of initializing the TUEVDataset with specific parameters and then calling its stat and info methods. This is useful for verifying dataset setup and content.

```python
if __name__ == "__main__":
    dataset = TUEVDataset(
        root="/srv/local/data/TUH/tuh_eeg_events/v2.0.0/edf",
        dev=True,
        refresh_cache=True,
    )
    dataset.stat()
    dataset.info()
    print(list(dataset.patients.items())[0])
```

--------------------------------

### FavMac Initialization and Calibration Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Demonstrates how to initialize the FavMac class with a trained model and target cost, then calibrate it using validation data. This setup is for controlling false positives in multi-label classification.

```python
>>> from pyhealth.calib.predictionset import FavMac
>>> from pyhealth.datasets import (
...     MIMIC3Dataset, get_dataloader,split_by_patient)
>>> from pyhealth.models import Transformer
>>> from pyhealth.tasks import drug_recommendation_mimic3_fn
>>> from pyhealth.trainer import get_metrics_fn
>>> base_dataset = MIMIC3Dataset(
...     root="/srv/scratch1/data/physionet.org/files/mimiciii/1.4",
...     tables=["DIAGNOSES_ICD", "PROCEDURES_ICD", "PRESCRIPTIONS"],
...     code_mapping={"NDC": ("ATC", {"target_kwargs": {"level": 3}})},
...     refresh_cache=False)
>>> sample_dataset = base_dataset.set_task(drug_recommendation_mimic3_fn)
>>> train_data, val_data, test_data = split_by_patient(sample_dataset, [0.6, 0.2, 0.2])
>>> model = Transformer(dataset=sample_dataset, feature_keys=["conditions", "procedures"],
...             label_key="drugs", mode="multilabel")
>>> # ... Train the model here ...
>>> # Try to control false positive to <=3
>>> cal_model = FavMac(model, target_cost=3, delta=None)
>>> cal_model.calibrate(cal_dataset=val_data)
```

--------------------------------

### Environment Setup and Imports

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/vision_embedding_tutorial.ipynb

Configures deterministic behavior and imports necessary libraries for the tutorial. Sets up the device to use CUDA if available, otherwise CPU.

```python
import os
import random
import tempfile
import shutil

import numpy as np
import torch
import torch.nn as nn
from PIL import Image

from pyhealth.datasets import create_sample_dataset
from pyhealth.datasets.splitter import split_by_patient
from pyhealth.datasets.utils import get_dataloader
from pyhealth.processors import ImageProcessor

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Environment Setup and PyHealth/PyTorch Check

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/medlink_mimic3.ipynb

Ensures the project root is in the system path and verifies the installation of PyTorch and PyHealth. This is a prerequisite for running other examples.

```python
import os
import sys

# Ensure project root is on sys.path when running from examples/
PROJECT_ROOT = os.path.abspath(os.path.join(os.getcwd(), ".."))
if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)

print("PROJECT_ROOT:", PROJECT_ROOT)

# Basic sanity check for torch and pyhealth
try:
    import torch
    print("PyTorch is installed")
except ImportError as e:
    raise RuntimeError(
        "PyTorch is not installed. Install it into your environment first " 
    ) from e

try:
    import pyhealth
    print("pyhealth is importable, version:", getattr(pyhealth, "__version__", "unknown"))
except ImportError as e:
    raise RuntimeError(
        "pyhealth is not importable."
    ) from e

# Core dataset + MedLink imports
from pyhealth.datasets import MIMIC3Dataset
from pyhealth.tasks import BaseTask
from pyhealth.models.medlink import (
    BM25Okapi,
    convert_to_ir_format,
    filter_by_candidates,
    generate_candidates,
    get_bm25_hard_negatives,
    get_eval_dataloader,
    get_train_dataloader,
    tvt_split,
)
from pyhealth.models.medlink.model import MedLink
from pyhealth.metrics import ranking_metrics_fn

```

--------------------------------

### Environment Setup and Library Imports

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/cnn_mimic4.ipynb

Configures deterministic behavior and imports necessary libraries for the tutorial. Sets random seeds for reproducibility and determines the appropriate device (CPU or GPU).

```python
import os
import random
from pathlib import Path

import numpy as np
import pandas as pd
import torch
from IPython.display import display

from pyhealth.datasets import MIMIC4Dataset
from pyhealth.datasets.splitter import split_by_patient
from pyhealth.datasets.utils import get_dataloader
from pyhealth.tasks.mortality_prediction import MortalityPredictionMIMIC4

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Initialize eICUDataset

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Example of initializing the eICUDataset with specified parameters. This setup is for loading and processing eICU data.

```python
dataset = eICUDataset(
        dataset_name=DATASET_NAME,
        root=ROOT,
        tables=TABLES,
        code_mapping=CODE_MAPPING,
        dev=DEV,
        refresh_cache=REFRESH_CACHE,
    )
```

--------------------------------

### Example Workflow: Model Training, Calibration, and Evaluation

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Demonstrates a typical workflow involving dataset loading, model initialization, training, calibration using KCal, and final evaluation with a Trainer.

```python
if __name__ == "__main__":

    from pyhealth.calib.calibration import KCal
    from pyhealth.datasets import (
        ISRUCDataset,
        get_dataloader,
        split_by_patient
    )
    from pyhealth.models import SparcNet
    from pyhealth.tasks import sleep_staging_isruc_fn

    sleep_ds = ISRUCDataset(
        root="/srv/local/data/trash",
        dev=True,
    ).set_task(sleep_staging_isruc_fn)
    train_data, val_data, test_data = split_by_patient(sleep_ds, [0.6, 0.2, 0.2])
    model = SparcNet(
        dataset=sleep_ds, feature_keys=["signal"], label_key="label", mode="multiclass"
    )
    # ... Train the model here ...
    # Calibrate
    cal_model = KCal(model)
    cal_model.calibrate(cal_dataset=val_data)
    # Evaluate
    from pyhealth.trainer import Trainer

    test_dl = get_dataloader(test_data, batch_size=32, shuffle=False)
    print(
        Trainer(model=cal_model, metrics=["cwECEt_adapt", "accuracy"])
        .evaluate(test_dl)
    )
```

--------------------------------

### Deepr Model Initialization and Usage Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/models/pyhealth.models.Deepr.md

Demonstrates how to initialize the Deepr model with a sample dataset, specify feature keys and label key, and perform a forward pass using a dataloader. This example requires PyTorch and PyHealth to be installed.

```python
>>> from pyhealth.datasets import SampleEHRDataset
>>> samples = [
...         {
...             "patient_id": "patient-0",
...             "visit_id": "visit-0",
...             "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
...             "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
...             "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
...             "list_list_vectors": [
...                 [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
...                 [[7.7, 8.5, 9.4]],
...             ],
...             "label": 1,
...         },
...         {
...             "patient_id": "patient-0",
...             "visit_id": "visit-1",
...             "list_codes": [
...                 "55154191800",
...                 "551541928",
...                 "55154192800",
...                 "705182798",
...                 "70518279800",
...             ],
...             "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
...             "list_list_codes": [["A04A", "B035", "C129"]],
...             "list_list_vectors": [
...                 [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
...             ],
...             "label": 0,
...         },
...     ]
>>> dataset = SampleEHRDataset(samples=samples, dataset_name="test")
>>>
>>> from pyhealth.models import Deepr
>>> model = Deepr(
...         dataset=dataset,
...         feature_keys=[
...             "list_list_codes",
...             "list_list_vectors",
...         ],
...         label_key="label",
...         mode="binary",
...     )
>>>
>>> from pyhealth.datasets import get_dataloader
>>> train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)
>>> data_batch = next(iter(train_loader))
>>>
>>> ret = model(**data_batch)
>>> print(ret)
{
    'loss': tensor(0.8908, device='cuda:0', grad_fn=<BinaryCrossEntropyWithLogitsBackward0>),
    'y_prob': tensor([[0.2295],
                [0.2665]], device='cuda:0', grad_fn=<SigmoidBackward0>),
    'y_true': tensor([[1.],
                [0.]], device='cuda:0'),
    'logit': tensor([[-1.2110],
                [-1.0126]], device='cuda:0', grad_fn=<AddmmBackward0>)
}
```

--------------------------------

### Environment Setup and Initialization

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/conformal_eeg/tuev_eeg_quickstart.ipynb

Seeds random generators, imports core dependencies, and detects the training device (CPU or CUDA).

```python
import random

import numpy as np
import torch

from pyhealth.datasets import TUEVDataset
from pyhealth.tasks import EEGEventsTUEV
from pyhealth.datasets.splitter import split_by_sample
from pyhealth.datasets.utils import get_dataloader

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Get Submodule Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/calib/pyhealth.calib.predictionset.md

Demonstrates how to retrieve a nested submodule using its fully-qualified string name. This method is efficient for checking submodule existence.

```text
A(
    (net_b): Module(
        (net_c): Module(
            (conv): Conv2d(16, 33, kernel_size=(3, 3), stride=(2, 2))
        )
        (linear): Linear(in_features=100, out_features=200, bias=True)
    )
)

get_submodule("net_b.linear")
get_submodule("net_b.net_c.conv")
```

--------------------------------

### Initialize Device and Print

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/text_embedding_tutorial.ipynb

Sets up the computation device (GPU or CPU) and prints the selected device. This is a common setup for PyTorch models.

```python
import torch
import warnings

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Getting Named Parameters

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/models/pyhealth.models.ContraWR.md

Provides an example of iterating over module parameters, yielding their names and the parameters. This snippet is skipped due to undefined variables in the original context.

```python
>>> # xdoctest: +SKIP("undefined vars")
>>> for name, param in self.named_parameters():
>>>     if name in ['bias']:
>>>         print(param.size())

```

--------------------------------

### Environment Setup and Device Selection

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/conformal_eeg/contrawr_tuev_eeg.ipynb

Initializes random seeds for reproducibility and selects the appropriate device (GPU or CPU) for computation. Ensure all necessary libraries are imported.

```python
import random

import numpy as np
import torch

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Environment Setup and Library Imports

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/agent_mimic4.ipynb

Configures deterministic behavior and imports necessary libraries for the tutorial, including PyHealth dataset and task modules, and PyTorch for deep learning.

```python
import os
import random
from pathlib import Path

import numpy as np
import torch
from IPython.display import display

from pyhealth.datasets import MIMIC4Dataset
from pyhealth.datasets.splitter import split_by_patient
from pyhealth.datasets.utils import get_dataloader
from pyhealth.tasks.mortality_prediction import MortalityPredictionMIMIC4

SEED = 42
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(SEED)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Running on device: {device}")
```

--------------------------------

### Get Submodule Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/models/pyhealth.models.MICRON.md

Demonstrates how to retrieve a nested submodule using its fully-qualified string name. This method is more efficient than iterating through all named modules for simple existence checks.

```python
a.get_submodule("net_b.linear")
a.get_submodule("net_b.net_c.conv")
```

--------------------------------

### Initialize SampleEHRDataset and RETAIN Model

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Demonstrates how to create a dataset from samples and initialize the RETAIN model with specified feature and label keys.

```python
from pyhealth.datasets import SampleEHRDataset
samples = [
        {
            "patient_id": "patient-0",
            "visit_id": "visit-0",
            "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
            "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
            "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
            "list_list_vectors": [
                [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
                [[7.7, 8.5, 9.4]],
            ],
            "label": 1,
        },
        {
            "patient_id": "patient-0",
            "visit_id": "visit-1",
            "list_codes": [
                "55154191800",
                "551541928",
                "55154192800",
                "705182798",
                "70518279800",
            ],
            "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
            "list_list_codes": [["A04A", "B035", "C129"]],
            "list_list_vectors": [
                [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
            ],
            "label": 0,
        },
    ]
dataset = SampleEHRDataset(samples=samples, dataset_name="test")

from pyhealth.models import RETAIN
model = RETAIN(
        dataset=dataset,
        feature_keys=[
            "list_codes",
            "list_vectors",
            "list_list_codes",
            "list_list_vectors",
        ],
        label_key="label",
        mode="binary",
    )
```

--------------------------------

### Setup Trainer with Default Settings

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/tutorials/tutorial_pyhealth_trainer.ipynb

Initialize the Trainer with a model and basic configuration. Logging is enabled by default, writing to a default output path and experiment name.

```python
trainer = Trainer(
    model=model,
    output_path="./output",
    exp_name="mortality_rnn_demo",
    enable_logging=True,
)

print(f"Training on device: {trainer.device}")
print(f"Logs written to:    {trainer.exp_path}")
```

--------------------------------

### Initialize SleepEDFDataset

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/datasets/pyhealth.datasets.SleepEDFDataset.md

Instantiate the SleepEDFDataset by providing the root directory of the dataset. This example shows how to initialize the dataset for the Sleep Cassette portion. You can then use the stat() and info() methods to get dataset statistics and information.

```python
from pyhealth.datasets import SleepEDFDataset
dataset = SleepEDFDataset(
        root="/srv/local/data/SLEEPEDF/sleep-edf-database-expanded-1.0.0/sleep-cassette",
    )
dataset.stat()
dataset.info()
```

--------------------------------

### HALO Model Initialization

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Demonstrates how to initialize the HALO model using a pre-configured processor and setting the device to CUDA if available, otherwise CPU. Includes optimizer definition.

```Python
import torch
device = "cuda" if torch.cuda.is_available() else "cpu"
model = HALO(
    n_ctx=processor.total_visit_size,
    total_vocab_size=processor.total_vocab_size,
    device=device
)

# define model optimizer:
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
```

--------------------------------

### Mortality Prediction Data Setup

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/mortality_prediction/timeseries_mimic4.ipynb

Initializes the dataset for mortality prediction, including setting the task, number of workers, and cache directory. It also provides information about multiprocessing start methods and potential issues.

```python
Label mortality vocab: {0: 0, 1: 1}
Processing samples and saving to ../benchmark_cache/mimic4_ihm_w_pre2...
Create an account on https://lightning.ai/ to optimize your data faster using multiple nodes and large machines.
Setting multiprocessing start_method to fork. Tip: Libraries relying on lock can hang with `fork`. To use `spawn` in notebooks, move your code to files and import it within the notebook.
Storing the files under /home/johnwu3/projects/PyHealth_Branch_Testing/PyHealth/benchmark_cache/mimic4_ihm_w_pre2
Setup started with fast_dev_run=False. 
Setup finished in 0.021 seconds. Found 1824 items to process.
Starting 4 workers with 1824 items. The progress bar is only updated when a worker finishes.
Workers are ready ! Starting data processing...
```

--------------------------------

### Quick Start: Using CheferRelevance

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/interpret/pyhealth.interpret.methods.chefer.md

Demonstrates how to initialize the CheferRelevance object, compute relevance scores for a test sample, and analyze the top relevant tokens. Ensure batch_size=1 for per-sample explanations and do not use within a torch.no_grad() context.

```python
from pyhealth.models import Transformer
from pyhealth.interpret.methods import CheferRelevance
from pyhealth.datasets import get_dataloader

# Assume you have a trained transformer model and dataset
model = Transformer(dataset=sample_dataset, ...)
# ... train the model ...

# Create interpretability object
relevance = CheferRelevance(model)

# Get a test sample (batch_size=1)
test_loader = get_dataloader(test_dataset, batch_size=1, shuffle=False)
batch = next(iter(test_loader))

# Compute relevance scores
scores = relevance.get_relevance_matrix(**batch)

# Analyze results
for feature_key, relevance_tensor in scores.items():
    print(f"{feature_key}: {relevance_tensor.shape}")
    top_tokens = relevance_tensor[0].topk(5).indices
    print(f
```

--------------------------------

### Initialize and Use StageNet Model

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/models/pyhealth.models.StageNet.md

Demonstrates how to initialize the StageNet model with a dataset, specify feature and time keys, and perform a forward pass to get predictions and loss. Requires dataset and dataloader setup.

```python
>>> from pyhealth.datasets import SampleEHRDataset
>>> samples = [
...     {
...         "patient_id": "patient-0",
...         "visit_id": "visit-0",
...         # "single_vector": [1, 2, 3],
...         "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
...         "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
...         "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
...         "list_list_vectors": [
...             [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
...             [[7.7, 8.5, 9.4]],
...         ],
...         "label": 1,
...         "list_vectors_time": [[0.0], [1.3]],
...         "list_codes_time": [[0.0], [2.0], [1.3]],
...         "list_list_codes_time": [[0.0], [1.5]],
...     },
...     {
...         "patient_id": "patient-0",
...         "visit_id": "visit-1",
...         # "single_vector": [1, 5, 8],
...         "list_codes": [
...             "55154191800",
...             "551541928",
...             "55154192800",
...             "705182798",
...             "70518279800",
...         ],
...         "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
...         "list_list_codes": [["A04A", "B035", "C129"]],
...         "list_list_vectors": [
...             [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
...         ],
...         "label": 0,
...         "list_vectors_time": [[0.0], [2.0], [1.0]],
...         "list_codes_time": [[0.0], [2.0], [1.3], [1.0], [2.0]],
...         "list_list_codes_time": [[0.0]],
...     },
... ]
>>>
>>> # dataset
>>> dataset = SampleEHRDataset(samples=samples, dataset_name="test")
>>>
>>> # data loader
>>> from pyhealth.datasets import get_dataloader
>>>
>>> train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)
>>>
>>> # model
>>> model = StageNet(
...     dataset=dataset,
...     feature_keys=[
...         "list_codes",
...         "list_vectors",
...         "list_list_codes",
...         # "list_list_vectors",
...     ],
...     time_keys=["list_codes_time", "list_vectors_time", "list_list_codes_time"],
...     label_key="label",
...     mode="binary",
... )
>>>
>>> from pyhealth.datasets import get_dataloader
>>> train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)
>>> data_batch = next(iter(train_loader))
>>>
>>> ret = model(**data_batch)
>>> print(ret)
{
    'loss': tensor(0.7111, grad_fn=<BinaryCrossEntropyWithLogitsBackward0>),
    'y_prob': tensor([[0.4815],
                [0.4991]], grad_fn=<SigmoidBackward0>),
    'y_true': tensor([[1.],
                [0.]]),
    'logit': tensor([[-0.0742],
                [-0.0038]], grad_fn=<AddmmBackward0>)
}
>>>
```

--------------------------------

### HALO Model Training Setup

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Instantiate the `HALOTrainer` with the required components for training the HALO model. Checkpoint files will be saved using the provided `checkpoint_name` and `checkpoint_path`.

```python
import os

# Assuming necessary imports for BaseEHRDataset, HALOProcessor, HALOModel, and optimizer
# from pyhealth.dataset import BaseEHRDataset
# from halo import HALOProcessor, HALOModel
# import torch

# Example placeholder for dataset, processor, model, and optimizer
# dataset = BaseEHRDataset(...) 
# processor = HALOProcessor(...)
# model = HALOModel(...)
# optimizer = torch.optim.Adam(model.parameters())

# checkpoint_name = "my_halo_checkpoint"
# checkpoint_path = "./checkpoints/halo"

# trainer = HALOTrainer(dataset=dataset, processor=processor, model=model, 
#                       optimizer=optimizer, checkpoint_name=checkpoint_name, 
#                       checkpoint_path=checkpoint_path)

# The loss function is present in the `HALOModel` class
# trainer.train()
```

--------------------------------

### Sample EHR Dataset and RNN Model Usage

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Demonstrates how to create a SampleEHRDataset, get a dataloader, instantiate an RNN model, and pass data through the model. Includes an example of performing a backward pass for loss calculation.

```Python
from pyhealth.datasets import SampleEHRDataset

samples = [
    {
        "patient_id": "patient-0",
        "visit_id": "visit-0",
        # "single_vector": [1, 2, 3],
        "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
        "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
        "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
        "list_list_vectors": [
            [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
            [[7.7, 8.5, 9.4]],
        ],
        "label": 1,
    },
    {
        "patient_id": "patient-0",
        "visit_id": "visit-1",
        # "single_vector": [1, 5, 8],
        "list_codes": [
            "55154191800",
            "551541928",
            "55154192800",
            "705182798",
            "70518279800",
        ],
        "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
        "list_list_codes": [["A04A", "B035", "C129"]],
        "list_list_vectors": [
            [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
        ],
        "label": 0,
    },
]

# dataset
dataset = SampleEHRDataset(samples=samples, dataset_name="test")

# data loader
from pyhealth.datasets import get_dataloader

train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)

# model
model = RNN(
    dataset=dataset,
    feature_keys=[
        "list_codes",
        "list_vectors",
        "list_list_codes",
        "list_list_vectors",
    ],
    label_key="label",
    mode="binary",
)

# data batch
data_batch = next(iter(train_loader))

# try the model
ret = model(**data_batch)
print(ret)

# try loss backward
ret["loss"].backward()
```

--------------------------------

### SCRIB Initialization and Setup

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

This snippet shows the initialization of the SCRIB (Set-classifier with Class-specific Risk Bounds) class, including setting up loss functions and parameters for coordinate descent optimization.

```python
"""
SCRIB: Set-classifier with Class-specific Risk Bounds

Implementation based on https://github.com/zlin7/scrib

"""

import time
from typing import Dict, Union

import numpy as np
import pandas as pd
import torch

from pyhealth.calib.base_classes import SetPredictor
from pyhealth.calib.utils import prepare_numpy_dataset
from pyhealth.models import BaseModel

from . import quicksearch as qs

OVERALL_LOSSFUNC = "overall"
CLASSPECIFIC_LOSSFUNC = "classspec"

__all__ = ["SCRIB"]


class _CoordDescent:
    def __init__(
        self,
        model_output,
        labels,
        rks,
        loss_func=OVERALL_LOSSFUNC,
        loss_kwargs=None,
        restart_n=1000,
        restart_range=0.1,
        init_range=None,
        verbose=False,
    ):
        self.N, self.K = model_output.shape

        # quantities useful for loss eval
        self.loss_name = loss_func
        if loss_kwargs is None:
            loss_kwargs = {}
        self.loss_kwargs = loss_kwargs
        if self.loss_name == OVERALL_LOSSFUNC:
            assert isinstance(rks, float)
        elif rks is not None:
            rks = np.asarray(rks)
        self.idx2rnk = np.asarray(
            pd.DataFrame(model_output).rank(ascending=True), np.int32
        )
        if np.min(self.idx2rnk) == 1:
            self.idx2rnk -= 1
        self.rnk2idx = np.asarray(np.argsort(model_output, axis=0), np.int32)
        if len(labels.shape) == 2:
            # one-hot -> class indices
            labels = np.argmax(labels, 1)
        self.labels = np.asarray(labels, np.int32)
        self.max_classes = np.argmax(model_output, 1)
        self.rks = rks

        self.model_output = model_output
        self.restart_n = restart_n
        self.restart_range = restart_range
        self.init_range = init_range or (int(np.ceil(self.N / 2)), self.N - 1)
        self.verbose = verbose

    def _search(self, ps):
        _search_fn = {
            CLASSPECIFIC_LOSSFUNC: qs.coord_desc_classspecific,
            OVERALL_LOSSFUNC: qs.coord_desc_overall,
        }[self.loss_name]
        return _search_fn(
            self.idx2rnk,
            self.rnk2idx,
            self.labels,
            self.max_classes,
            ps,
            self.rks,
            **self.loss_kwargs,
        )

    def _loss_eval(self, ps):
        _loss_fn = {
            CLASSPECIFIC_LOSSFUNC: qs.loss_classspecific,
            OVERALL_LOSSFUNC: qs.loss_overall,
        }[self.loss_name]
        return _loss_fn(
            self.idx2rnk,
            self.rnk2idx,
            self.labels,
            self.max_classes,
            ps,
            self.rks,
            **self.loss_kwargs,
        )

    def _p2t(self, p):
        # Translate ranks to thresholds
        return [self.model_output[self.rnk2idx[p[k], k], k] for k in range(self.K)]

    def _sample_new_loc(self, old_p, restart_range=0.1):
        diff = np.random.uniform(-restart_range, restart_range, self.K)
        new_p = old_p.copy()
        for k in range(self.K):
            new_p[k] = max(min(int(new_p[k] + diff[k] * self.N), self.N - 1), 0)
        return new_p

    def search_once(self, seed=7):
        def print_(s):
            if self.verbose:
                print(s)

        np.random.seed(seed)

```

--------------------------------

### Load MIMIC-III Dataset

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/tcn_mimic3_codes.ipynb

Loads the MIMIC-III dataset from a specified root URL, including diagnosis, procedures, and prescriptions tables. Development mode is enabled for smaller data subsets. Use this to get started with the dataset.

```python
from pyhealth.datasets import MIMIC3Dataset

dataset = MIMIC3Dataset(
    root="https://storage.googleapis.com/pyhealth/Synthetic_MIMIC-III",
    tables=["DIAGNOSES_ICD", "PROCEDURES_ICD", "PRESCRIPTIONS"],
    dev=True,
)
dataset.stats()
```

--------------------------------

### Initialize GRASP Model and Perform Forward Pass

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/docs/api/models/pyhealth.models.GRASP.md

Demonstrates how to initialize the GRASP model with a sample dataset and feature keys, and then perform a forward pass using a data batch to get predictions and loss. Ensure the dataset and dataloader are set up correctly.

```python
>>> from pyhealth.datasets import SampleEHRDataset
>>> samples = [
...         {
...             "patient_id": "patient-0",
...             "visit_id": "visit-0",
...             "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
...             "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
...             "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
...             "list_list_vectors": [
...                 [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
...                 [[7.7, 8.5, 9.4]],
...             ],
...             "demographic": [0.0, 2.0, 1.5],
...             "label": 1,
...         },
...         {
...             "patient_id": "patient-0",
...             "visit_id": "visit-1",
...             "list_codes": [
...                 "55154191800",
...                 "551541928",
...                 "55154192800",
...                 "705182798",
...                 "70518279800",
...             ],
...             "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
...             "list_list_codes": [["A04A", "B035", "C129"]],
...             "list_list_vectors": [
...                 [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
...             ],
...             "demographic": [0.0, 2.0, 1.5],
...             "label": 0,
...         },
...     ]
>>> dataset = SampleEHRDataset(samples=samples, dataset_name="test")
>>>
>>> from pyhealth.models import GRASP
>>> model = GRASP(
...         dataset=dataset,
...         feature_keys=[
...             "list_codes",
...             "list_vectors",
...             "list_list_codes",
...             "list_list_vectors",
...         ],
...         label_key="label",
...         static_key="demographic",
...         use_embedding=[True, False, True, False],
...         mode="binary"
...     )
>>>
>>> from pyhealth.datasets import get_dataloader
>>> train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)
>>> data_batch = next(iter(train_loader))
>>>
>>> ret = model(**data_batch)
>>> print(ret)
{
    'loss': tensor(0.6896, grad_fn=<BinaryCrossEntropyWithLogitsBackward0>),
    'y_prob': tensor([[0.4983],
                [0.4947]], grad_fn=<SigmoidBackward0>),
    'y_true': tensor([[1.],
                [0.]]),
    'logit': tensor([[-0.0070],
                [-0.0213]], grad_fn=<AddmmBackward0>)
}
>>>
```

--------------------------------

### FavMac Calibration and Inference Example

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Demonstrates how to initialize and use the FavMac class for controlling false positives in a drug recommendation task using MIMIC-III data. It includes model training setup, calibration, and evaluation.

```python
from pyhealth.calib.predictionset import FavMac
from pyhealth.datasets import (MIMIC3Dataset, get_dataloader,split_by_patient)
from pyhealth.models import Transformer
from pyhealth.tasks import drug_recommendation_mimic3_fn
from pyhealth.trainer import get_metrics_fn
base_dataset = MIMIC3Dataset(
    root="/srv/scratch1/data/physionet.org/files/mimiciii/1.4",
    tables=["DIAGNOSES_ICD", "PROCEDURES_ICD", "PRESCRIPTIONS"],
    code_mapping={"NDC": ("ATC", {"target_kwargs": {"level": 3}})
    },
    refresh_cache=False)
sample_dataset = base_dataset.set_task(drug_recommendation_mimic3_fn)
train_data, val_data, test_data = split_by_patient(sample_dataset, [0.6, 0.2, 0.2])
model = Transformer(dataset=sample_dataset, feature_keys=["conditions", "procedures"],
            label_key="drugs", mode="multilabel")
# ... Train the model here ...
# Try to control false positive to <=3
cal_model = FavMac(model, target_cost=3, delta=None)
cal_model.calibrate(cal_dataset=val_data)
# Evaluate
from pyhealth.trainer import Trainer
test_dl = get_dataloader(test_data, batch_size=32, shuffle=False)
y_true_all, y_prob_all, _, extra_output = Trainer(model=cal_model).inference(
test_dl, additional_outputs=["y_predset"])
print(get_metrics_fn(cal_model.mode)(
    y_true_all, y_prob_all, metrics=['tp', 'fp'],
```

--------------------------------

### Initialize and Use OMOPDataset

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Demonstrates how to initialize an OMOPDataset with specified tables and then retrieve its statistics and information. Assumes the dataset is located at the provided root path.

```python
from pyhealth.datasets import OMOPDataset
dataset = OMOPDataset(
        root="/srv/local/data/zw12/pyhealth/raw_data/synpuf1k_omop_cdm_5.2.2",
        tables=["condition_occurrence", "procedure_occurrence", "drug_exposure", "measurement"],
    )
dataset.stat()
dataset.info()
```

--------------------------------

### Get ICD Events within a Time Interval

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/tutorials/orig_tutorial_pyhealth_datasets.ipynb

Fetches 'procedures_icd' events for a patient that occurred after a specific start time, derived from the patient's first admission event. Requires the dataset to be loaded and admissions to be available.

```python
admissions = dataset.get_patient('1').get_events("admissions")
start = admissions[0].timestamp
procedures = dataset.get_patient('1').get_events("procedures_icd", start=start)
print(procedures)
```

--------------------------------

### Train Model with Basic Configuration

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/examples/tutorials/tutorial_pyhealth_trainer.ipynb

Start the training process with specified dataloaders, epochs, and optimizer. Gradient clipping and monitoring can be configured.

```python
trainer.train(
    train_dataloader,
    val_dataloader     = None,
    test_dataloader    = None,
    epochs             = 5,
    optimizer_class    = torch.optim.Adam,
    optimizer_params   = {"lr": 1e-3},
    weight_decay       = 0.0,
    max_grad_norm      = None,        # gradient clipping (e.g., 1.0)
    monitor            = None,        # metric name to track for best model
    monitor_criterion  = "max",       # "max" or "min"
    load_best_model_at_last = True,   # restore best weights after training
    patience           = None,        # early stopping patience in epochs
)
```

--------------------------------

### Process EEG Data and Get Patient Information

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth.txt

Illustrates the process_EEG_data method within TUEVDataset, which organizes EEG files and patient information. The example shows how to retrieve patient data, including signal and label files, for training and evaluation sets.

```python
import os

# Assuming TUEVDataset class is defined elsewhere and self.root, self.filepath, self.dev are initialized

# get all file names
all_files = {}

train_files = os.listdir(os.path.join(self.root, "train/"))
for id in train_files:
    if id != ".DS_Store":
        all_files["0_{}".format(id)] = [name for name in os.listdir(os.path.join(self.root, "train/", id)) if name.endswith(".edf")]

eval_files = os.listdir(os.path.join(self.root, "eval/"))
for id in eval_files:
    if id != ".DS_Store":
        all_files["1_{}".format(id)] = [name for name in os.listdir(os.path.join(self.root, "eval/", id)) if name.endswith(".edf")]

# get all patient ids
patient_ids = list(set(list(all_files.keys())))

if self.dev:
    patient_ids = patient_ids[:20]
    # print(patient_ids)

# get patient to record maps
#    - key: pid:
#    - value: [{"load_from_path": None, "patient_id": None, "signal_file": None, "label_file": None, "save_to_path": None}, ...]
patients = {
    pid: []
    for pid in patient_ids
}
   
for pid in patient_ids:
    split = "train" if pid.split("_")[0] == "0" else "eval"
    id = pid.split("_")[1]

    patient_visits = all_files[pid]
    
    for visit in patient_visits:
        if split == "train":
            visit_id = visit.strip(".edf").split("_")[1]
        else:
            visit_id = visit.strip(".edf")
            
        patients[pid].append({
            "load_from_path": os.path.join(self.root, split, id),
            "patient_id": pid,
            "visit_id": visit_id,
            "signal_file": visit,
            "label_file": visit,
            "save_to_path": self.filepath,
        })

return patients
```

--------------------------------

### PyHealth Sample EHR Dataset and Dataloader Setup

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

This example demonstrates how to create a sample EHR dataset and set up a dataloader for training. It defines sample patient data with various features and labels, then uses `get_dataloader` to prepare data for model input.

```python
if __name__ == "__main__":
    from pyhealth.datasets import SampleEHRDataset

    samples = [
        {
            "patient_id": "patient-0",
            "visit_id": "visit-0",
            "single_vector": [1, 2, 3],
            "list_codes": ["505800458", "50580045810", "50580045811"],  # NDC
            "list_vectors": [[1.0, 2.55, 3.4], [4.1, 5.5, 6.0]],
            "list_list_codes": [["A05B", "A05C", "A06A"], ["A11D", "A11E"]],  # ATC-4
            "list_list_vectors": [
                [[1.8, 2.25, 3.41], [4.50, 5.9, 6.0]],
                [[7.7, 8.5, 9.4]],
            ],
            "label": 1,
        },
        {
            "patient_id": "patient-0",
            "visit_id": "visit-1",
            "single_vector": [1, 5, 8],
            "list_codes": [
                "55154191800",
                "551541928",
                "55154192800",
                "705182798",
                "70518279800",
            ],
            "list_vectors": [[1.4, 3.2, 3.5], [4.1, 5.9, 1.7], [4.5, 5.9, 1.7]],
            "list_list_codes": [["A04A", "B035", "C129"]],
            "list_list_vectors": [
                [[1.0, 2.8, 3.3], [4.9, 5.0, 6.6], [7.7, 8.4, 1.3], [7.7, 8.4, 1.3]],
            ],
            "label": 0,
        },
    ]

    # dataset
    dataset = SampleEHRDataset(samples=samples, dataset_name="test")

    # data loader
    from pyhealth.datasets import get_dataloader

    train_loader = get_dataloader(dataset, batch_size=2, shuffle=True)

    # model
    model = Deepr(
        dataset=dataset,
        # feature_keys=["procedures"],
        feature_keys=["list_list_codes", "list_list_vectors"],
        label_key="label",
        mode="binary",

```

--------------------------------

### Install PyHealth

Source: https://github.com/sunlabuiuc/pyhealth/blob/master/chat-assistant/corpus/pyhealth-code.txt

Installs the PyHealth library in editable mode.

```bash
pip install -e .
```
Tokens:
Raw