diff --git a/docs/api/models.rst b/docs/api/models.rst index e881e7894..769d9ccce 100644 --- a/docs/api/models.rst +++ b/docs/api/models.rst @@ -247,6 +247,7 @@ API Reference models/pyhealth.models.CorGAN models/pyhealth.models.Synthea models/pyhealth.models.SDOH + models/pyhealth.models.SequenceEmbeddingModel models/pyhealth.models.VisionEmbeddingModel models/pyhealth.models.TextEmbedding models/pyhealth.models.BIOT diff --git a/docs/api/models/pyhealth.models.SequenceEmbeddingModel.rst b/docs/api/models/pyhealth.models.SequenceEmbeddingModel.rst new file mode 100644 index 000000000..74ca799da --- /dev/null +++ b/docs/api/models/pyhealth.models.SequenceEmbeddingModel.rst @@ -0,0 +1,18 @@ +pyhealth.models.SequenceEmbeddingModel +====================================== + +Embeddings for the packed records of ``sequence_v2`` +(:class:`~pyhealth.processors.SequenceProcessorV2`) fields. Each level's +records get one embedding per token namespace and a projection of their +values, concatenated; each level keeps its offsets and times. + +.. autoclass:: pyhealth.models.SequenceEmbeddingModel + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: pyhealth.models.LevelEmbedding + :members: + +.. autoclass:: pyhealth.models.PretrainedEmbedding + :members: diff --git a/pyhealth/models/__init__.py b/pyhealth/models/__init__.py index a9e6f90b2..5c430d584 100644 --- a/pyhealth/models/__init__.py +++ b/pyhealth/models/__init__.py @@ -8,6 +8,11 @@ from .contrawr import ContraWR, ResBlock2D from .deepr import Deepr, DeeprLayer from .embedding import EmbeddingModel +from .sequence_embedding import ( + LevelEmbedding as LevelEmbedding, + PretrainedEmbedding as PretrainedEmbedding, + SequenceEmbeddingModel as SequenceEmbeddingModel, +) from .gamenet import GAMENet, GAMENetLayer from .jamba_ehr import JambaEHR, JambaLayer from .logistic_regression import LogisticRegression diff --git a/pyhealth/models/sequence_embedding.py b/pyhealth/models/sequence_embedding.py new file mode 100644 index 000000000..32f174db5 --- /dev/null +++ b/pyhealth/models/sequence_embedding.py @@ -0,0 +1,425 @@ +"""Embeddings for the packed records of :class:`SequenceProcessorV2` fields.""" + +import math +from collections.abc import Mapping, Sequence +from typing import Any, Literal, Required, TypedDict, cast + +import torch +from torch import nn + +from ..datasets import SampleDataset +from ..processors import SequenceProcessorV2 +from ..processors.sequence_processor_v2 import FieldKey +from .base_model import BaseModel +from .embedding import init_embedding_with_pretrained + +MaskMode = Literal["ignore", "concat"] + + +class LevelEmbedding(TypedDict, total=False): + """Embedding sizes of one level; a plain ``dict`` with type hints. + + A key is required exactly when the level has that role, and ``{}`` is a + level without tokens or values. + + Keys: + tokens: One size per token namespace (``K`` sizes). Each namespace has + its own ``nn.Embedding``; the record's token embedding is their + concatenation. + values: The output size of the linear layer over the values. + + Examples: + >>> labs: LevelEmbedding = {"tokens": (8, 8), "values": 16} + >>> sum(labs["tokens"]) + labs["values"] # the record's embedding size + 32 + """ + + tokens: Sequence[int] + values: int + + +class PretrainedEmbedding(TypedDict, total=False): + """A pretrained text-vector file for one token namespace; a plain ``dict``. + + Keys: + field: The V2 field. + level: The level of the field. + namespace: The token namespace of the level. + path: A text file with one token and its vector per line (e.g. GloVe); + the vector size must be the namespace's embedding size. Tokens + not in the file keep their random initialization. + freeze: Whether to freeze the embedding. Default False. + normalize: Reserved for future work; only None. + + Examples: + >>> glove: PretrainedEmbedding = { + ... "field": "notes", "level": 0, "namespace": 0, + ... "path": "glove.6B.100d.txt", "freeze": True, + ... } + >>> glove["freeze"] + True + """ + + field: Required[str] + level: Required[int] + namespace: Required[int] + path: Required[str] + freeze: bool + normalize: None + + +def _broadcast(x: torch.Tensor, spatial: tuple[int, ...]) -> torch.Tensor: + """``[N, D]`` -> ``[N, *spatial, D]``; an ``x`` with spatial axes is kept.""" + if x.dim() == 2 + len(spatial): + return x + return x.reshape(x.shape[0], *([1] * len(spatial)), x.shape[-1]).expand( + x.shape[0], *spatial, x.shape[-1] + ) + + +class _FieldEmbedding(nn.Module): + """The embedding layers of one V2 field, one level after another.""" + + def __init__( + self, + field: str, + processor: SequenceProcessorV2, + dims: Sequence[LevelEmbedding], + flatten_spatial: Sequence[bool], + mask: MaskMode, + ): + super().__init__() + self.batch_schema = processor.batch_schema() + self.mask = mask + masked = {level for level, role in self.batch_schema if role == "mask"} + self.depth = processor.depth + vocab_size, value_dim, spatial = ( + processor.vocab_size(), + processor.value_dim(), + processor.spatial(), + ) + if not isinstance(dims, (list, tuple)) or len(dims) != self.depth + 1: + raise ValueError( + f"embedding_dim['{field}'] must have one dict per level " + f"({self.depth + 1}), got {dims!r}." + ) + if len(flatten_spatial) != self.depth + 1: + raise ValueError( + f"flatten_spatial['{field}'] must have one bool per level " + f"({self.depth + 1}), got {flatten_spatial!r}." + ) + + # Per level: the token embeddings ("token_"), the values projection + # ("value_proj"), and how the values are laid out before it. (Not + # "values": that would shadow ModuleDict.values().) + self.levels = nn.ModuleList() + self.sizes: list[int] = [] # the record embedding size per level + self.spatial_axes: list[tuple[int, ...]] = [] # value axes kept apart + for level, (dim, flat) in enumerate(zip(dims, flatten_spatial)): + where = f"embedding_dim['{field}'][{level}]" + if not isinstance(dim, dict) or not dim.keys() <= {"tokens", "values"}: + raise ValueError(f"{where} must be a dict of 'tokens' / 'values'.") + layers = nn.ModuleDict() + size = 0 + namespaces = vocab_size[level] + if namespaces is None: + if "tokens" in dim: + raise ValueError(f"{where}: level {level} has no tokens.") + else: + sizes = dim.get("tokens") + if ( + not isinstance(sizes, (list, tuple)) + or len(sizes) != len(namespaces) + or not all(isinstance(s, int) and s > 0 for s in sizes) + ): + raise ValueError( + f"{where}['tokens'] must be {len(namespaces)} positive " + f"sizes, one per token namespace, got {sizes!r}." + ) + for k, (vocab, width) in enumerate(zip(namespaces, sizes)): + layers[f"token_{k}"] = nn.Embedding(vocab, width, padding_idx=0) + size += sum(sizes) + + shape, flags = value_dim[level], spatial[level] + kept: tuple[int, ...] = () + if shape is None: + if "values" in dim: + raise ValueError(f"{where}: level {level} has no values.") + if not flat: + raise ValueError( + f"flatten_spatial['{field}'][{level}]: level {level} has " + "no values." + ) + else: + width = dim.get("values") + if not isinstance(width, int) or width <= 0: + raise ValueError( + f"{where}['values'] must be a positive size, got {width!r}." + ) + assert flags is not None # a level with values has spatial flags + if not flat: + kept = tuple(axis for axis, on in enumerate(flags) if on) + if not kept: + raise ValueError( + f"flatten_spatial['{field}'][{level}] is False, but no " + "value axis of the level is spatial." + ) + features = math.prod( + size for axis, size in enumerate(shape) if axis not in kept + ) + # With "concat", the mask's features join the values'. + concat = level in masked and mask == "concat" + layers["value_proj"] = nn.Linear((2 if concat else 1) * features, width) + size += width + self.levels.append(layers) + self.sizes.append(size) + self.spatial_axes.append(kept) + + # Spatial shape of each level's embeddings, e.g. (16,) for kept ECG + # samples; a level's token embedding is broadcast over it. + self.spatial_shapes = [ + tuple(value_dim[level][axis] for axis in kept) # type: ignore[index] + for level, kept in enumerate(self.spatial_axes) + ] + + def layer(self, level: int, name: str) -> nn.Module: + """One layer of a level: ``token_`` or ``value_proj``.""" + return cast(nn.ModuleDict, self.levels[level])[name] + + def embed_level( + self, level: int, fields: Mapping[FieldKey, torch.Tensor] + ) -> torch.Tensor | None: + """The ``[N, *spatial, D]`` embeddings of one level's records, or None.""" + parts = [] + tokens = fields.get((level, "tokens")) + if tokens is not None: + tokens = tokens.unsqueeze(1) if tokens.dim() == 1 else tokens + parts.append( + torch.cat( + [ + self.layer(level, f"token_{k}")(tokens[:, k]) + for k in range(tokens.shape[1]) + ], + dim=-1, + ) + ) + values = fields.get((level, "values")) + if values is not None: + kept = self.spatial_axes[level] + others = [1 + a for a in range(values.dim() - 1) if a not in kept] + + def features(x: torch.Tensor) -> torch.Tensor: + # [N, *S] -> [N, *S_kept, prod(S_others)], to project per position. + x = x.permute(0, *(1 + a for a in kept), *others) + return x.reshape( + *x.shape[: 1 + len(kept)], math.prod(x.shape[1 + len(kept) :]) + ) + + x = features(values) + mask = fields.get((level, "mask")) + if mask is not None and self.mask == "concat": + x = torch.cat([x, features(mask.float())], dim=-1) + parts.append(self.layer(level, "value_proj")(x)) + if not parts: + return None + spatial = self.spatial_shapes[level] + return torch.cat([_broadcast(p, spatial) for p in parts], dim=-1) + + def forward(self, inputs: Sequence[torch.Tensor]) -> dict[Any, Any]: + fields = dict(zip(self.batch_schema, inputs)) + levels = { + level: { + "embedding": self.embed_level(level, fields), + "offsets": fields[(level, "offsets")], + "time": fields.get((level, "time")), + } + for level in range(self.depth + 1) + } + return levels + + +class SequenceEmbeddingModel(BaseModel): + """Embeds the packed records of every :class:`SequenceProcessorV2` field. + + Like :class:`~pyhealth.models.EmbeddingModel`, it is a building block for + other models: :meth:`forward` maps each V2 field of a batch (the tuple of + :meth:`SequenceProcessorV2.collate`) to embeddings. Fields of other + processors are ignored and left out of the output. + + A record's embedding at level ``l`` is the concatenation of one + ``nn.Embedding`` per token namespace (```` embeds to zeros) and a + linear projection of its values, so its size is + ``D_l = sum(tokens) + values``. ``time`` is passed through unchanged. + + Args: + dataset: The dataset, with fitted processors. + embedding_dim: Per V2 field, one :class:`LevelEmbedding` per level, + innermost first. Every V2 field must be listed. + flatten_spatial: Per V2 field, one bool per level (default all True). + True flattens all value axes before the projection, giving + ``[N, values]``. False flattens only the non-spatial axes and + projects at each position of the spatial ones, giving + ``[N, *spatial, values]``; the level's token embedding is + broadcast over those positions. + pretrain: :class:`PretrainedEmbedding` entries that initialize token + namespaces from text-vector files. + mask: How levels with a ``mask`` (``missing_indicator``) use it. + + - ``"ignore"`` (default): the mask is not used. + - ``"concat"``: the values and the mask are projected together, + ``Linear(cat(values, mask))``. + + Examples: + >>> from pyhealth.datasets import create_sample_dataset + >>> samples = [ + ... {"patient_id": "p0", "visits": {"children": [{"tokens": ["A", "B"]}, {"tokens": ["C"]}]}, "label": 0}, + ... {"patient_id": "p1", "visits": {"children": [{"tokens": ["C"]}]}, "label": 1}, + ... ] + >>> levels = [{"ordering": "set", "tokens": 1}, {"ordering": "sequence"}] + >>> dataset = create_sample_dataset( + ... samples, {"visits": ("sequence_v2", {"levels": levels})}, {"label": "binary"} + ... ) + >>> model = SequenceEmbeddingModel(dataset, embedding_dim={"visits": [{"tokens": (8,)}, {}]}) + >>> proc = dataset.input_processors["visits"] + >>> batch = proc.collate([dataset[0]["visits"], dataset[1]["visits"]]) + >>> out = model({"visits": batch})["visits"] + >>> out[0]["embedding"].shape, out[0]["offsets"].tolist(), out[1]["offsets"].tolist() + (torch.Size([4, 8]), [0, 2, 3, 4], [0, 2, 3]) + """ + + def __init__( + self, + dataset: SampleDataset, + embedding_dim: Mapping[str, Sequence[LevelEmbedding]], + flatten_spatial: Mapping[str, Sequence[bool]] | None = None, + pretrain: Sequence[PretrainedEmbedding] | None = None, + mask: MaskMode = "ignore", + ): + super().__init__(dataset) + if mask not in ("ignore", "concat"): + raise ValueError(f"mask must be 'ignore' or 'concat', got {mask!r}.") + processors = { + field: processor + for field, processor in dataset.input_processors.items() + if isinstance(processor, SequenceProcessorV2) + } + flatten_spatial = dict(flatten_spatial or {}) + for name, config in ( + ("embedding_dim", embedding_dim), + ("flatten_spatial", flatten_spatial), + ): + unknown = set(config) - processors.keys() + if unknown: + raise ValueError( + f"{name} names {sorted(unknown)}, which are not " + f"SequenceProcessorV2 fields ({sorted(processors)})." + ) + missing = processors.keys() - set(embedding_dim) + if missing: + raise ValueError(f"embedding_dim has no entry for {sorted(missing)}.") + self.embedding_layers = nn.ModuleDict( + { + field: _FieldEmbedding( + field, + processor, + embedding_dim[field], + flatten_spatial.get(field, [True] * (processor.depth + 1)), + mask, + ) + for field, processor in processors.items() + } + ) + seen = set() + for entry in pretrain or (): + self._load_pretrained(entry, processors, seen) + + def embedding_dim_per_level(self, field: str) -> tuple[int, ...]: + """The record embedding size ``D_l`` of each level of a V2 field. + + Examples: + >>> from pyhealth.datasets import create_sample_dataset + >>> labs = {"tokens": [("Glucose", "mg/dL")], "values": [[1.0]]} + >>> samples = [{"patient_id": f"p{i}", "labs": labs, "y": i} for i in range(2)] + >>> dataset = create_sample_dataset( + ... samples, {"labs": ("sequence_v2", {"levels": [{"tokens": 2, "values": [1]}]})}, {"y": "binary"} + ... ) + >>> model = SequenceEmbeddingModel(dataset, embedding_dim={"labs": [{"tokens": (8, 4), "values": 16}]}) + >>> model.embedding_dim_per_level("labs") + (28,) + """ + return tuple(cast(_FieldEmbedding, self.embedding_layers[field]).sizes) + + def _load_pretrained( + self, + entry: PretrainedEmbedding, + processors: Mapping[str, SequenceProcessorV2], + seen: set[tuple[str, int, int]], + ) -> None: + if ( + not isinstance(entry, dict) + or not {"field", "level", "namespace", "path"} <= entry.keys() + ): + raise ValueError( + f"pretrain entries need 'field', 'level', 'namespace' and 'path', " + f"got {entry!r}." + ) + if entry.get("normalize") is not None: + raise NotImplementedError( + "pretrain 'normalize' is reserved for future work and only " + "accepts None for now." + ) + field, level, namespace = entry["field"], entry["level"], entry["namespace"] + if field not in processors: + raise ValueError(f"pretrain: '{field}' is not a SequenceProcessorV2 field.") + vocab_size = processors[field].vocab_size() + if not ( + isinstance(level, int) + and 0 <= level < len(vocab_size) + and vocab_size[level] is not None + and isinstance(namespace, int) + and 0 <= namespace < len(vocab_size[level]) # type: ignore[arg-type] + ): + raise ValueError( + f"pretrain: '{field}' has no token namespace {namespace} at level " + f"{level}." + ) + if (field, level, namespace) in seen: + raise ValueError( + f"pretrain: '{field}' level {level} namespace {namespace} is listed " + "twice." + ) + seen.add((field, level, namespace)) + layers = cast(_FieldEmbedding, self.embedding_layers[field]) + embedding = layers.layer(level, f"token_{namespace}") + assert isinstance(embedding, nn.Embedding) + init_embedding_with_pretrained( + embedding, + processors[field].vocab(level, namespace), + entry["path"], + embedding.embedding_dim, + freeze=entry.get("freeze", False), + ) + + # BaseModel.forward(**kwargs) is the predictor contract (loss, y_prob, ...); + # like EmbeddingModel, this building block embeds given inputs instead. + def forward( # type: ignore[override] + self, inputs: Mapping[str, Any] + ) -> dict[str, dict[Any, Any]]: + """Embeds the V2 fields of a batch. + + Args: + inputs: Batch fields, e.g. a batch of + :func:`~pyhealth.datasets.get_dataloader`. Only the V2 fields + are read; others are ignored. + + Returns: + Per V2 field, per level: ``{"embedding": [N_l, *spatial, D_l] + or None, "offsets": the level's offsets (the batch offsets at + the top level), "time": [N_l] or None}``. + """ + device = self.device + return { + field: layers(tuple(tensor.to(device) for tensor in inputs[field])) + for field, layers in self.embedding_layers.items() + if field in inputs + } diff --git a/tests/core/test_sequence_embedding.py b/tests/core/test_sequence_embedding.py new file mode 100644 index 000000000..8260b7fc8 --- /dev/null +++ b/tests/core/test_sequence_embedding.py @@ -0,0 +1,355 @@ +import os +import tempfile +import unittest +from typing import Any, cast + +import torch + +from pyhealth.datasets import create_sample_dataset, get_dataloader +from pyhealth.models import LevelEmbedding, SequenceEmbeddingModel +from pyhealth.processors import LevelConfig, SequenceProcessorV2 + +NAN = float("nan") +# stays (series) -> labs (series) with tokens (test, unit) and a value +LABS: list[LevelConfig] = [ + {"ordering": "series", "tokens": 2, "values": [1]}, + {"ordering": "series"}, +] +# stays -> ECG windows [leads, samples], spatial along the samples, with a rhythm token +ECG: list[LevelConfig] = [ + {"ordering": "series", "tokens": 1, "values": [2, 4], "spatial": [False, True]}, + {"ordering": "sequence"}, +] + + +def _dataset(**fields: Any): + schema = { + name: ("sequence_v2", {"levels": levels}) + for name, (levels, _) in fields.items() + } + samples = [ + { + "patient_id": f"p{i}", + **{name: records[i] for name, (_, records) in fields.items()}, + "y": i % 2, + } + for i in range(len(next(iter(fields.values()))[1])) + ] + return create_sample_dataset(samples, schema, {"y": "binary"}) + + +def _labs(*stays: list[tuple[float, str, str, float]]) -> dict: + return { + "time": [10.0 * i for i in range(len(stays))], + "children": [ + { + "time": [row[0] for row in stay], + "tokens": [(row[1], row[2]) for row in stay], + "values": [[row[3]] for row in stay], + } + for stay in stays + ], + } + + +LAB_RECORDS = [ + _labs( + [(0.0, "Glucose", "mg/dL", 100.0), (1.0, "WBC", "10^3/uL", 7.0)], + [(2.0, "Glucose", "mg/dL", 90.0)], + ), + _labs([(0.0, "WBC", "10^3/uL", 8.0)]), +] +LAB_DIMS: dict[str, list[LevelEmbedding]] = { + "labs": [{"tokens": (4, 2), "values": 3}, {}] +} + + +def _layer(model: SequenceEmbeddingModel, field: str, level: int, name: str) -> Any: + """One embedding layer of a field's level (``token_``, ``value_proj``, ...).""" + return cast(Any, model.embedding_layers[field]).layer(level, name) + + +def _batch(dataset, field: str): + proc = dataset.input_processors[field] + assert isinstance(proc, SequenceProcessorV2) + return proc.collate([dataset[i][field] for i in range(len(dataset))]) + + +class TestConfiguration(unittest.TestCase): + def setUp(self): + self.dataset = _dataset(labs=(LABS, LAB_RECORDS)) + + def test_rejects_bad_configurations(self): + bad: list[dict[str, Any]] = [ + {"embedding_dim": {}}, # labs missing + {"embedding_dim": {**LAB_DIMS, "other": [{}]}}, + {"embedding_dim": {"labs": [{"tokens": (4, 2), "values": 3}]}}, # one level + {"embedding_dim": {"labs": [{"tokens": (4,), "values": 3}, {}]}}, # K = 2 + {"embedding_dim": {"labs": [{"tokens": (4, 0), "values": 3}, {}]}}, + {"embedding_dim": {"labs": [{"tokens": (4, 2)}, {}]}}, # values missing + { + "embedding_dim": { + "labs": [{"tokens": (4, 2), "values": 3}, {"values": 2}] + } + }, + { + "embedding_dim": { + "labs": [{"tokens": (4, 2), "values": 3}, {"tokens": (2,)}] + } + }, # level 1 has no tokens + { + "embedding_dim": { + "labs": [{"tokens": (4, 2), "values": 3, "time": 1}, {}] + } + }, + { + "embedding_dim": LAB_DIMS, + "flatten_spatial": {"labs": [False, True]}, + }, # no spatial + { + "embedding_dim": LAB_DIMS, + "flatten_spatial": {"labs": [True, False]}, + }, # level 1 has no values + {"embedding_dim": LAB_DIMS, "flatten_spatial": {"labs": [True]}}, + {"embedding_dim": LAB_DIMS, "flatten_spatial": {"other": [True]}}, + ] + for kwargs in bad: + with self.subTest(kwargs=kwargs), self.assertRaises(ValueError): + SequenceEmbeddingModel(self.dataset, **kwargs) + + def test_embedding_dim_per_level(self): + model = SequenceEmbeddingModel(self.dataset, embedding_dim=LAB_DIMS) + self.assertEqual(model.embedding_dim_per_level("labs"), (4 + 2 + 3, 0)) + + def test_ignores_fields_of_other_processors(self): + samples = [ + {"patient_id": "p0", "labs": LAB_RECORDS[0], "age": [60.0], "y": 0}, + {"patient_id": "p1", "labs": LAB_RECORDS[1], "age": [70.0], "y": 1}, + ] + dataset = create_sample_dataset( + samples, + {"labs": ("sequence_v2", {"levels": LABS}), "age": "tensor"}, + {"y": "binary"}, + ) + model = SequenceEmbeddingModel(dataset, embedding_dim=LAB_DIMS) + with self.assertRaises(ValueError): # "age" is not a V2 field + SequenceEmbeddingModel(dataset, embedding_dim={**LAB_DIMS, "age": [{}]}) + batch = next(iter(get_dataloader(dataset, batch_size=2))) + self.assertEqual(set(model(batch)), {"labs"}) + + +class TestLevels(unittest.TestCase): + def test_all_empty_numeric_records(self) -> None: + dataset = _dataset(x=([ + {"ordering": "set", "values": [2, 4], "spatial": [False, True]}, + ], [{"values": []}, {"values": []}])) + for flatten in (True, False): + model = SequenceEmbeddingModel( + dataset, embedding_dim={"x": [{"values": 3}]}, + flatten_spatial={"x": [flatten]}, + ) + out = model({"x": _batch(dataset, "x")})["x"][0] + self.assertEqual( + out["embedding"].shape, (0, 3) if flatten else (0, 4, 3) + ) + self.assertEqual(out["offsets"].tolist(), [0, 0, 0]) + out["embedding"].sum().backward() + + def test_level_embeddings_offsets_and_time(self): + dataset = _dataset(labs=(LABS, LAB_RECORDS)) + model = SequenceEmbeddingModel(dataset, embedding_dim=LAB_DIMS) + batch = _batch(dataset, "labs") + out = model({"labs": batch})["labs"] + self.assertEqual(out[0]["embedding"].shape, (4, 4 + 2 + 3)) + self.assertEqual(out[0]["offsets"].tolist(), [0, 2, 3, 4]) + self.assertEqual(out[0]["time"].tolist(), [0.0, 1.0, 2.0, 0.0]) + self.assertIsNone(out[1]["embedding"]) # level 1 has only time + self.assertEqual(out[1]["offsets"].tolist(), [0, 2, 3]) # batch offsets + self.assertEqual(out[1]["time"].tolist(), [0.0, 10.0, 0.0]) + + # The embedding is the namespaces' embeddings and the projected values. + tokens, values = batch[0], batch[1] + expected = torch.cat( + [ + _layer(model, "labs", 0, "token_0")(tokens[:, 0]), + _layer(model, "labs", 0, "token_1")(tokens[:, 1]), + _layer(model, "labs", 0, "value_proj")(values), + ], + dim=1, + ) + torch.testing.assert_close(out[0]["embedding"], expected) + + def test_pad_embeds_to_zero(self): + levels: list[LevelConfig] = [ + { + "ordering": "series", + "tokens": 1, + "resample": {"step": 1.0, "method": "last"}, + }, + {}, + ] + records = [{"children": [{"time": [0.0, 2.0], "tokens": ["A", "B"]}]}] * 2 + dataset = _dataset(codes=(levels, records)) + dims: dict[str, list[LevelEmbedding]] = {"codes": [{"tokens": (5,)}, {}]} + model = SequenceEmbeddingModel(dataset, embedding_dim=dims) + out = model({"codes": _batch(dataset, "codes")})["codes"][0]["embedding"] + self.assertEqual(out.shape, (6, 5)) + self.assertTrue(torch.all(out[1] == 0)) # step 1 has no record: + self.assertTrue(torch.any(out[0] != 0)) + + def test_flatten_spatial(self): + windows = [ + { + "children": [ + { + "time": [0.0], + "tokens": ["sinus"], + "values": torch.rand(1, 2, 4).tolist(), + } + ] + }, + { + "children": [ + { + "time": [0.0, 1.0], + "tokens": ["afib", "sinus"], + "values": torch.rand(2, 2, 4).tolist(), + } + ] + }, + ] + dataset = _dataset(ecg=(ECG, windows)) + dims: dict[str, list[LevelEmbedding]] = { + "ecg": [{"tokens": (3,), "values": 5}, {}] + } + batch = _batch(dataset, "ecg") + + flat = SequenceEmbeddingModel(dataset, embedding_dim=dims) + self.assertEqual(flat({"ecg": batch})["ecg"][0]["embedding"].shape, (3, 3 + 5)) + + model = SequenceEmbeddingModel( + dataset, embedding_dim=dims, flatten_spatial={"ecg": [False, True]} + ) + out = model({"ecg": batch})["ecg"][0]["embedding"] + self.assertEqual(out.shape, (3, 4, 3 + 5)) # one embedding per sample position + tokens, values = batch[0], batch[1] + # The rhythm token is broadcast over the positions... + token = _layer(model, "ecg", 0, "token_0")(tokens) + torch.testing.assert_close(out[:, :, :3], token[:, None, :].expand(3, 4, 3)) + # ...and the leads are projected at each position. + torch.testing.assert_close( + out[:, :, 3:], _layer(model, "ecg", 0, "value_proj")(values.transpose(1, 2)) + ) + + +class TestMask(unittest.TestCase): + def setUp(self): + levels: list[LevelConfig] = [ + { + "ordering": "series", + "values": [2], + "impute": {"method": "zero"}, + "missing_indicator": True, + }, + {}, + ] + records = [ + {"children": [{"time": [0.0, 1.0], "values": [[1.0, NAN], [2.0, 3.0]]}]}, + {"children": [{"time": [0.0], "values": [[NAN, 4.0]]}]}, + ] + self.dataset = _dataset(vitals=(levels, records)) + self.dims: dict[str, list[LevelEmbedding]] = {"vitals": [{"values": 6}, {}]} + self.batch = _batch(self.dataset, "vitals") + self.values, self.mask = self.batch[0], self.batch[1] + + def embed(self, mode: Any): + model = SequenceEmbeddingModel(self.dataset, embedding_dim=self.dims, mask=mode) + return model, model({"vitals": self.batch})["vitals"][0]["embedding"] + + def test_rejects_unknown_mode(self): + with self.assertRaises(ValueError): + unknown: Any = "gated" # not supported (yet) + SequenceEmbeddingModel(self.dataset, embedding_dim=self.dims, mask=unknown) + + def test_ignore(self): + model, out = self.embed("ignore") + torch.testing.assert_close( + out, _layer(model, "vitals", 0, "value_proj")(self.values) + ) + + def test_concat(self): + model, out = self.embed("concat") + self.assertEqual(_layer(model, "vitals", 0, "value_proj").in_features, 4) + x = torch.cat([self.values, self.mask.float()], dim=1) + torch.testing.assert_close(out, _layer(model, "vitals", 0, "value_proj")(x)) + + +class TestPretrain(unittest.TestCase): + def setUp(self): + self.dataset = _dataset(labs=(LABS, LAB_RECORDS)) + handle, self.path = tempfile.mkstemp(suffix=".txt") + with os.fdopen(handle, "w") as f: + f.write("Glucose 1 2 3 4\nunused 9 9 9 9\n") + self.addCleanup(os.remove, self.path) + + def entry(self, **changes: Any) -> Any: + return { + "field": "labs", + "level": 0, + "namespace": 0, + "path": self.path, + **changes, + } + + def test_loads_and_freezes(self): + model = SequenceEmbeddingModel( + self.dataset, embedding_dim=LAB_DIMS, pretrain=[self.entry(freeze=True)] + ) + embedding = _layer(model, "labs", 0, "token_0") + proc = self.dataset.input_processors["labs"] + assert isinstance(proc, SequenceProcessorV2) + index = proc.vocab(0, 0)["Glucose"] + self.assertEqual(embedding.weight[index].tolist(), [1.0, 2.0, 3.0, 4.0]) + self.assertTrue(torch.all(embedding.weight[0] == 0)) # + self.assertFalse(embedding.weight.requires_grad) + + def test_rejects_bad_entries(self): + bad: list[Any] = [ + [{"field": "labs", "level": 0, "path": self.path}], # no namespace + [self.entry(field="other")], + [self.entry(level=1)], # level 1 has no tokens + [self.entry(namespace=2)], + [self.entry(), self.entry()], + ] + for pretrain in bad: + with self.subTest(pretrain=pretrain), self.assertRaises(ValueError): + SequenceEmbeddingModel( + self.dataset, embedding_dim=LAB_DIMS, pretrain=pretrain + ) + with self.assertRaises(NotImplementedError): + SequenceEmbeddingModel( + self.dataset, + embedding_dim=LAB_DIMS, + pretrain=[self.entry(normalize="l2")], + ) + with self.assertRaises(FileNotFoundError): + SequenceEmbeddingModel( + self.dataset, + embedding_dim=LAB_DIMS, + pretrain=[self.entry(path=self.path + ".missing")], + ) + + +class TestTraining(unittest.TestCase): + def test_dataloader_batch_and_gradients(self): + dataset = _dataset(labs=(LABS, LAB_RECORDS)) + model = SequenceEmbeddingModel(dataset, embedding_dim=LAB_DIMS) + batch = next(iter(get_dataloader(dataset, batch_size=2))) + out = model(batch)["labs"] + out[0]["embedding"].sum().backward() + self.assertIsNotNone(_layer(model, "labs", 0, "value_proj").weight.grad) + self.assertIsNotNone(_layer(model, "labs", 0, "token_1").weight.grad) + + +if __name__ == "__main__": + unittest.main()