From 456355dd7f18a06bfbee408fa5ce082a70380825 Mon Sep 17 00:00:00 2001 From: Logiquo Date: Tue, 27 Oct 2026 02:43:21 -0400 Subject: [PATCH 1/4] Add timeseries label processors and normalization options Label processors: timeseries_binary and timeseries_regression (TimeseriesLabelProcessor) place per-step labels on the same uniform time grid as TimeseriesProcessor and return (value, mask), with mask 0 for steps without a label; they feed the timeseries_* model modes. Gridding is shared through sample_on_grid. Preprocessing options, defaults unchanged: - TimeseriesProcessor: normalization ("zscore" or "minmax", statistics learned in fit from the observed values) applied before imputation, and missing-value indicators appended. With zscore and forward_fill, values before the first observation become 0, the training mean. - TensorProcessor: the same normalization, missing-value indicators and impute_strategy="zero". Both set learns_statistics when normalizing, so fitting without a training split warns. The shared helpers live in processors/utils.py. --- pyhealth/processors/__init__.py | 10 +- pyhealth/processors/tensor_processor.py | 69 ++++++- pyhealth/processors/timeseries_processor.py | 217 +++++++++++++++++--- pyhealth/processors/utils.py | 70 +++++++ tests/core/test_per_step_labels.py | 122 +++++++++++ tests/core/test_processor_normalization.py | 95 +++++++++ 6 files changed, 553 insertions(+), 30 deletions(-) create mode 100644 pyhealth/processors/utils.py create mode 100644 tests/core/test_per_step_labels.py create mode 100644 tests/core/test_processor_normalization.py diff --git a/pyhealth/processors/__init__.py b/pyhealth/processors/__init__.py index 5fd0a12ba..b0738b05a 100644 --- a/pyhealth/processors/__init__.py +++ b/pyhealth/processors/__init__.py @@ -44,7 +44,12 @@ def get_processor(name: str): ) from .tensor_processor import TensorProcessor from .text_processor import TextProcessor -from .timeseries_processor import TimeseriesProcessor +from .timeseries_processor import ( + TimeseriesBinaryLabelProcessor, + TimeseriesLabelProcessor, + TimeseriesProcessor, + TimeseriesRegressionLabelProcessor, +) from .time_image_processor import TimeImageProcessor from .graph_processor import GraphProcessor from .audio_processor import AudioProcessor @@ -77,7 +82,10 @@ def get_processor(name: str): "TensorProcessor", "TemporalTimeseriesProcessor", "TextProcessor", + "TimeseriesBinaryLabelProcessor", + "TimeseriesLabelProcessor", "TimeseriesProcessor", + "TimeseriesRegressionLabelProcessor", "TimeImageProcessor", "GraphProcessor", "AudioProcessor", diff --git a/pyhealth/processors/tensor_processor.py b/pyhealth/processors/tensor_processor.py index 3c15cffe7..1bb30af8d 100644 --- a/pyhealth/processors/tensor_processor.py +++ b/pyhealth/processors/tensor_processor.py @@ -1,9 +1,10 @@ -from typing import Any, Dict, Iterable, Optional +from typing import Any, Dict, Iterable, List, Literal, Optional import torch from . import register_processor from .base_processor import FeatureProcessor +from .utils import NORMALIZATIONS, Normalization, fit_normalization @register_processor("tensor") @@ -16,6 +17,11 @@ class TensorProcessor(FeatureProcessor): Processing: - Convert input directly to torch.Tensor using torch.tensor() + - Optionally, along the last dimension: normalization with statistics + learned in ``fit()`` from the observed (non-NaN) values, + missing-value indicators (1 where the value is NaN) appended, and + NaN filled with 0, in that order. After ``"zscore"`` + normalization, 0 is the training mean. Output: - torch.Tensor with appropriate shape and dtype @@ -31,11 +37,18 @@ class TensorProcessor(FeatureProcessor): """ stores_tensor = True + # Defaults for processors pickled before these options existed. + normalization: Normalization = None + missing_indicators: bool = False + impute_strategy: Optional[Literal["zero"]] = None def __init__( self, dtype: torch.dtype = torch.float32, spatial_dims: Optional[tuple[bool, ...]] = None, + normalization: Normalization = None, + missing_indicators: bool = False, + impute_strategy: Optional[Literal["zero"]] = None, ): """ Initialize the TensorProcessor. @@ -45,11 +58,27 @@ def __init__( Default is torch.float32. spatial_dims: Tuple of booleans indicating which dimensions are spatial. If None, defaults to all False. Default is None. + normalization: ``None``, ``"zscore"`` (zero mean, unit variance) + or ``"minmax"`` (scaled to [0, 1] on the fitted range), + per entry of the last dimension. Default is None. + missing_indicators: Whether to append a missing-value indicator + per entry of the last dimension. Default is False. + impute_strategy: ``None`` (keep NaN) or ``"zero"``. Default is None. """ + if normalization not in NORMALIZATIONS: + raise ValueError(f"normalization must be one of {NORMALIZATIONS}") + if impute_strategy not in (None, "zero"): + raise ValueError("impute_strategy must be None or 'zero'") self.dtype = dtype self._n_dim = None self._feature_dim = None self._spatial_dims = spatial_dims + self.normalization = normalization + self.missing_indicators = missing_indicators + self.impute_strategy = impute_strategy + self.learns_statistics = normalization is not None + self.center: List[float] = [] + self.scale: List[float] = [] def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None: """Infer n_dim and the feature width from the first valid sample. @@ -70,6 +99,18 @@ def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None: self._feature_dim = tensor.shape[-1] if tensor.dim() > 0 else 1 break + if self.normalization is not None: + self.center, self.scale = fit_normalization( + ( + torch.as_tensor(sample[field], dtype=torch.float64) + .reshape(-1, self._feature_dim) + .numpy() + for sample in samples + if sample.get(field) is not None + ), + self.normalization, + ) + def process(self, value: Any) -> torch.Tensor: """ Process a numerical value or list into a torch.Tensor. @@ -84,8 +125,23 @@ def process(self, value: Any) -> torch.Tensor: # which can trigger a UserWarning. If value is already a tensor, # return a detached clone cast to the requested dtype. if isinstance(value, torch.Tensor): - return value.detach().clone().to(dtype=self.dtype) - return torch.tensor(value, dtype=self.dtype) + tensor = value.detach().clone().to(dtype=self.dtype) + else: + tensor = torch.tensor(value, dtype=self.dtype) + + if self.normalization is not None: + center = torch.tensor(self.center, dtype=self.dtype) + scale = torch.tensor(self.scale, dtype=self.dtype) + tensor = (tensor - center) / scale + if self.missing_indicators: + if tensor.dim() == 0: + tensor = tensor.reshape(1) + indicators = torch.isnan(tensor).to(self.dtype) + if self.impute_strategy == "zero": + tensor = torch.nan_to_num(tensor, nan=0.0) + if self.missing_indicators: + tensor = torch.cat([tensor, indicators], dim=-1) + return tensor def size(self) -> int | None: """ @@ -93,9 +149,12 @@ def size(self) -> int | None: Returns: The width of the last dimension (1 for scalars) seen in ``fit()``, - or None before fitting. + doubled with missing-value indicators, or None before fitting. """ - return getattr(self, "_feature_dim", None) + feature_dim = getattr(self, "_feature_dim", None) + if feature_dim is None or not self.missing_indicators: + return feature_dim + return feature_dim * 2 def is_token(self) -> bool: """Whether the output tensor represents discrete token indices, inferred from dtype. diff --git a/pyhealth/processors/timeseries_processor.py b/pyhealth/processors/timeseries_processor.py index 3d1e253a4..c1335378a 100644 --- a/pyhealth/processors/timeseries_processor.py +++ b/pyhealth/processors/timeseries_processor.py @@ -6,6 +6,38 @@ from . import register_processor from .base_processor import FeatureProcessor +from .utils import NORMALIZATIONS, Normalization, fit_normalization + + +def sample_on_grid( + timestamps: List[datetime], values: np.ndarray, sampling_rate: timedelta +) -> np.ndarray: + """Places values on a uniform time grid starting at the first timestamp. + + Args: + timestamps: Time of each row of ``values``. + values: Array of shape ``(T,)`` or ``(T, F)``. + sampling_rate: Interval of the grid. + + Returns: + Array of shape ``(S, F)``, NaN where the grid has no value. + """ + values = np.asarray(values, dtype=float) + if values.ndim == 1: + values = values[:, None] + num_features = values.shape[1] + + start_time = timestamps[0] + end_time = timestamps[-1] + total_steps = int((end_time - start_time) / sampling_rate) + 1 + sampled_values = np.full((total_steps, num_features), np.nan) + + # Map original timestamps to indices in the sampled grid + for t, v in zip(timestamps, values): + idx = int((t - start_time) / sampling_rate) + if 0 <= idx < total_steps: + sampled_values[idx] = v + return sampled_values @register_processor("timeseries") @@ -19,10 +51,27 @@ class TimeseriesProcessor(FeatureProcessor): Processing: 1. Uniform sampling at fixed intervals. - 2. Imputation for missing values. + 2. Optional normalization, with statistics learned in ``fit()`` from + the observed (non-missing) sampled values. + 3. Optional missing-value indicators, one per feature, 1 where the + value was missing before imputation. + 4. Imputation for missing values. + + With ``normalization="zscore"`` and ``impute_strategy="forward_fill"``, + values missing before the first observation are filled with 0, which is + the training mean after normalization. Output: - - torch.Tensor of shape (S, F), where S is the number of sampled time steps. + - torch.Tensor of shape (S, F), or (S, 2F) with missing-value + indicators appended, where S is the number of sampled time steps. + + Args: + sampling_rate: Interval of the sampled time steps. + impute_strategy: ``"forward_fill"`` (values before the first + observation are filled with 0) or ``"zero"``. + normalization: ``None``, ``"zscore"`` (zero mean, unit variance) or + ``"minmax"`` (scaled to [0, 1] on the fitted range). + missing_indicators: Whether to append the missing-value indicators. Examples: >>> from datetime import datetime, timedelta @@ -36,19 +85,34 @@ class TimeseriesProcessor(FeatureProcessor): """ stores_tensor = True + # Defaults for processors pickled before these options existed. + normalization: Normalization = None + missing_indicators: bool = False def __init__( self, sampling_rate: timedelta = timedelta(hours=1), impute_strategy: str = "forward_fill", + normalization: Normalization = None, + missing_indicators: bool = False, ): + if normalization not in NORMALIZATIONS: + raise ValueError(f"normalization must be one of {NORMALIZATIONS}") # Configurable sampling rate and imputation method self.sampling_rate = sampling_rate self.impute_strategy = impute_strategy + self.normalization = normalization + self.missing_indicators = missing_indicators + self.learns_statistics = normalization is not None self.n_features = None + self.center: List[float] = [] + self.scale: List[float] = [] def fit(self, samples: Any, field: str) -> None: - """Fit the processor by determining n_features from the first valid sample. + """Fit the processor. + + Determines n_features from the first valid sample and, with + ``normalization``, learns the per-feature statistics from all samples. Args: samples: Iterable of sample dictionaries. @@ -66,32 +130,42 @@ def fit(self, samples: Any, field: str) -> None: self.n_features = 1 break + if self.normalization is not None: + self.center, self.scale = fit_normalization( + ( + self._sample(*sample[field]) + for sample in samples + if sample.get(field) is not None and len(sample[field][0]) > 0 + ), + self.normalization, + ) + + def _sample( + self, timestamps: List[datetime], values: np.ndarray + ) -> np.ndarray: + return sample_on_grid(timestamps, values, self.sampling_rate) + def process(self, value: Tuple[List[datetime], np.ndarray]) -> torch.Tensor: timestamps, values = value if len(timestamps) == 0: raise ValueError("Timestamps list is empty.") - values = np.asarray(values) - num_features = values.shape[1] - # Step 1: Uniform sampling - start_time = timestamps[0] - end_time = timestamps[-1] - total_steps = int((end_time - start_time) / self.sampling_rate) + 1 - - sampled_times = [ - start_time + i * self.sampling_rate for i in range(total_steps) - ] - sampled_values = np.full((total_steps, num_features), np.nan) - - # Map original timestamps to indices in the sampled grid - for t, v in zip(timestamps, values): - idx = int((t - start_time) / self.sampling_rate) - if 0 <= idx < total_steps: - sampled_values[idx] = v - - # Step 2: Imputation + sampled_values = self._sample(timestamps, values) + total_steps, num_features = sampled_values.shape + + # Step 2: Normalization + if self.normalization is not None: + sampled_values = ( + sampled_values - np.asarray(self.center) + ) / np.asarray(self.scale) + + # Step 3: Missing-value indicators, before imputation + if self.missing_indicators: + indicators = np.isnan(sampled_values).astype(float) + + # Step 4: Imputation if self.impute_strategy == "forward_fill": for f in range(num_features): last_value = 0.0 @@ -108,11 +182,15 @@ def process(self, value: Tuple[List[datetime], np.ndarray]) -> torch.Tensor: if self.n_features is None: self.n_features = sampled_values.shape[1] + if self.missing_indicators: + sampled_values = np.concatenate([sampled_values, indicators], axis=1) return torch.tensor(sampled_values, dtype=torch.float) def size(self): # Size equals number of features, unknown until first process - return self.n_features + if self.n_features is None: + return None + return self.n_features * (2 if self.missing_indicators else 1) def is_token(self) -> bool: """Time series values are continuous, not discrete tokens.""" @@ -132,5 +210,96 @@ def spatial(self) -> tuple[bool, ...]: def __repr__(self): return ( f"TimeSeriesProcessor(sampling_rate={self.sampling_rate}, " - f"impute_strategy='{self.impute_strategy}')" + f"impute_strategy='{self.impute_strategy}', " + f"normalization={self.normalization!r}, " + f"missing_indicators={self.missing_indicators})" + ) + + +class TimeseriesLabelProcessor(FeatureProcessor): + """Base processor for labels given per time step. + + The label is a time series ``(timestamps, values)``, placed on the same + kind of uniform grid as :class:`TimeseriesProcessor`. With the same first + timestamp and ``sampling_rate`` as the input time series, each label lines + up with one input step. Steps without a label (NaN or not on the grid) + are masked out and never imputed. + + Output: + - ``(value, mask)``: two float tensors of shape (S,), where ``mask`` + is 1 for steps with a label. Padding added when batching has mask 0. + + The registered names ``"timeseries_binary"`` and + ``"timeseries_regression"`` are model modes of their own, so modules that + assume one prediction per sample (calibration, prediction sets, + interpretability) do not treat these labels as ``"binary"`` or + ``"regression"``. + + Args: + sampling_rate: Interval of the sampled time steps. + """ + + def __init__(self, sampling_rate: timedelta = timedelta(hours=1)): + self.sampling_rate = sampling_rate + + def process( + self, value: Tuple[List[datetime], np.ndarray] + ) -> Tuple[torch.Tensor, torch.Tensor]: + timestamps, labels = value + if len(timestamps) == 0: + raise ValueError("Timestamps list is empty.") + sampled = sample_on_grid(timestamps, labels, self.sampling_rate) + if sampled.shape[1] != 1: + raise ValueError("Expected one label per time step.") + sampled = sampled[:, 0] + mask = ~np.isnan(sampled) + self._check(sampled[mask]) + return ( + torch.tensor(np.where(mask, sampled, 0.0), dtype=torch.float), + torch.tensor(mask, dtype=torch.float), ) + + def _check(self, labels: np.ndarray) -> None: + pass + + def size(self) -> int: + return 1 + + def is_token(self) -> bool: + return False + + def schema(self) -> tuple[str, ...]: + return ("value", "mask") + + def dim(self) -> tuple[int, ...]: + return (1, 1) + + def spatial(self) -> tuple[bool, ...]: + return (True,) + + def __repr__(self): + return f"{type(self).__name__}(sampling_rate={self.sampling_rate})" + + +@register_processor("timeseries_binary") +class TimeseriesBinaryLabelProcessor(TimeseriesLabelProcessor): + """Binary (0/1) label per time step, see :class:`TimeseriesLabelProcessor`. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor + >>> t0 = datetime(2020, 1, 1) + >>> times = [t0 + timedelta(hours=h) for h in range(3)] + >>> TimeseriesBinaryLabelProcessor().process((times, [0, float("nan"), 1])) + (tensor([0., 0., 1.]), tensor([1., 0., 1.])) + """ + + def _check(self, labels: np.ndarray) -> None: + if not np.isin(labels, (0.0, 1.0)).all(): + raise ValueError("Binary labels must be 0 or 1.") + + +@register_processor("timeseries_regression") +class TimeseriesRegressionLabelProcessor(TimeseriesLabelProcessor): + """Real-valued label per time step, see :class:`TimeseriesLabelProcessor`.""" + diff --git a/pyhealth/processors/utils.py b/pyhealth/processors/utils.py new file mode 100644 index 000000000..bd9d0bbb6 --- /dev/null +++ b/pyhealth/processors/utils.py @@ -0,0 +1,70 @@ +"""Utilities shared by processors.""" + +from typing import Iterable, List, Literal, Optional, Tuple + +import numpy as np + +Normalization = Optional[Literal["zscore", "minmax"]] +NORMALIZATIONS = (None, "zscore", "minmax") + + +def fit_normalization( + arrays: Iterable[np.ndarray], method: Normalization +) -> Tuple[List[float], List[float]]: + """Learns per-feature ``(center, scale)`` from the observed values. + + Missing values (``NaN``) are ignored. ``"zscore"`` uses the mean and the + population standard deviation, ``"minmax"`` the minimum and the range. A + feature that is never observed or constant gets a scale of 1. + + Args: + arrays: Arrays of shape ``(N, F)``, streamed one at a time. + method: ``"zscore"`` or ``"minmax"``. + + Returns: + The center and scale of each feature, as lists of floats. + """ + count = total_mean = total_m2 = low = high = None + for array in arrays: + array = np.asarray(array, dtype=np.float64) + if count is None: + n_features = array.shape[1] + count = np.zeros(n_features) + total_mean = np.zeros(n_features) + total_m2 = np.zeros(n_features) + low = np.full(n_features, np.inf) + high = np.full(n_features, -np.inf) + observed = ~np.isnan(array) + n = observed.sum(axis=0) + if not n.any(): + continue + with np.errstate(invalid="ignore", divide="ignore"): + mean = np.where(n > 0, np.nansum(array, axis=0) / n, 0.0) + m2 = np.nansum((array - mean) ** 2, axis=0) + # Chan et al. parallel update of the running mean and squared deviations. + combined = count + n + delta = mean - total_mean + with np.errstate(invalid="ignore", divide="ignore"): + total_mean = np.where( + combined > 0, total_mean + delta * n / combined, 0.0 + ) + total_m2 = total_m2 + m2 + np.where( + combined > 0, delta**2 * count * n / combined, 0.0 + ) + count = combined + low = np.fmin(low, np.nanmin(np.where(observed, array, np.inf), axis=0)) + high = np.fmax(high, np.nanmax(np.where(observed, array, -np.inf), axis=0)) + + if count is None: + return [], [] + if method == "zscore": + center = total_mean + with np.errstate(invalid="ignore", divide="ignore"): + scale = np.sqrt(np.where(count > 0, total_m2 / count, 0.0)) + elif method == "minmax": + center = np.where(count > 0, low, 0.0) + scale = np.where(count > 0, high - low, 0.0) + else: + raise ValueError(f"normalization must be one of {NORMALIZATIONS}") + scale = np.where(scale > 0, scale, 1.0) + return center.tolist(), scale.tolist() diff --git a/tests/core/test_per_step_labels.py b/tests/core/test_per_step_labels.py new file mode 100644 index 000000000..bfeb3f65d --- /dev/null +++ b/tests/core/test_per_step_labels.py @@ -0,0 +1,122 @@ +import unittest +from datetime import datetime, timedelta + +import numpy as np +import torch + +from pyhealth.datasets import create_sample_dataset, get_dataloader +from pyhealth.models import RNN +from pyhealth.processors import ( + TimeseriesBinaryLabelProcessor, + TimeseriesRegressionLabelProcessor, +) +from pyhealth.trainer import Trainer + +T0 = datetime(2020, 1, 1) +NAN = float("nan") + + +def _times(n): + return [T0 + timedelta(hours=h) for h in range(n)] + + +def _samples(label_fn): + rng = np.random.default_rng(0) + samples = [] + for i, n in enumerate([5, 3, 6, 4]): + samples.append( + { + "patient_id": f"p{i}", + "dynamic": (_times(n), rng.normal(size=(n, 2))), + "static": [float(i), 1.0], + "label": (_times(n), [label_fn(i, h) for h in range(n)]), + } + ) + return samples + + +def _dataset(output="timeseries_binary", label_fn=lambda i, h: float((i + h) % 2)): + return create_sample_dataset( + samples=_samples(label_fn), + input_schema={"dynamic": "timeseries", "static": "tensor"}, + output_schema={"label": output}, + dataset_name="per_step", + ) + + +class TestTimeseriesLabelProcessors(unittest.TestCase): + def test_value_and_mask(self): + value, mask = TimeseriesBinaryLabelProcessor().process( + (_times(4), [0, NAN, 1, 1]) + ) + np.testing.assert_array_equal(value.numpy(), [0, 0, 1, 1]) + np.testing.assert_array_equal(mask.numpy(), [1, 0, 1, 1]) + + def test_grid_follows_sampling_rate(self): + times = [T0, T0 + timedelta(hours=3)] + value, mask = TimeseriesRegressionLabelProcessor().process((times, [2.5, 4.0])) + np.testing.assert_array_equal(value.numpy(), [2.5, 0, 0, 4.0]) + np.testing.assert_array_equal(mask.numpy(), [1, 0, 0, 1]) + + def test_binary_rejects_other_values(self): + with self.assertRaises(ValueError): + TimeseriesBinaryLabelProcessor().process((_times(2), [0, 2])) + + +class TestPerStepModel(unittest.TestCase): + def test_modes_are_separate_from_binary_and_regression(self): + model = RNN(dataset=_dataset("timeseries_binary")) + self.assertEqual(model.mode, "timeseries_binary") + model = RNN(dataset=_dataset("timeseries_regression")) + self.assertEqual(model.mode, "timeseries_regression") + + def test_loss_ignores_unlabeled_and_padded_steps(self): + # Stay 1 has an unlabeled step and is padded; changing what the + # model predicts there must not change the loss. + dataset = _dataset(label_fn=lambda i, h: NAN if (i, h) == (1, 0) else float(h % 2)) + model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval() + batch = next(iter(get_dataloader(dataset, batch_size=4))) + with torch.no_grad(): + out = model(**batch) + value, mask = out["y_true"] + self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1)) + self.assertEqual(int(mask.sum()), 17) # 5 + 3 + 6 + 4 steps, minus one + + logits = out["logit"].clone() + loss_fn = model.get_loss_function() + logits[mask == 0] = 1e3 + torch.testing.assert_close(loss_fn(logits, (value, mask)), out["loss"]) + + def test_predictions_do_not_see_future_steps(self): + dataset = _dataset() + model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval() + batch = next(iter(get_dataloader(dataset, batch_size=1))) + with torch.no_grad(): + before = model(**batch)["y_prob"][0, :, 0] + batch["dynamic"] = batch["dynamic"].clone() + batch["dynamic"][:, -1] += 100.0 + after = model(**batch)["y_prob"][0, :, 0] + torch.testing.assert_close(before[:-1], after[:-1]) + self.assertFalse(torch.allclose(before[-1], after[-1])) + + def test_bidirectional_is_rejected(self): + with self.assertRaises(ValueError): + RNN(dataset=_dataset(), bidirectional=True) + + def test_trainer_scores_labeled_steps_per_sample(self): + dataset = _dataset() + model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8) + trainer = Trainer(model=model, metrics=["roc_auc"], enable_logging=False) + loader = get_dataloader(dataset, batch_size=3) + y_true, y_prob, _ = trainer.inference(loader) + self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6]) + self.assertEqual([len(p) for p in y_prob], [len(y) for y in y_true]) + self.assertIn("roc_auc", trainer.evaluate(loader)) + + dataset = _dataset("timeseries_regression") + model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8) + trainer = Trainer(model=model, metrics=["mae"], enable_logging=False) + self.assertIn("mae", trainer.evaluate(get_dataloader(dataset, batch_size=3))) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/core/test_processor_normalization.py b/tests/core/test_processor_normalization.py new file mode 100644 index 000000000..99dec0a70 --- /dev/null +++ b/tests/core/test_processor_normalization.py @@ -0,0 +1,95 @@ +import unittest +from datetime import datetime, timedelta + +import numpy as np +import torch + +from pyhealth.processors import TensorProcessor, TimeseriesProcessor +from pyhealth.processors.utils import fit_normalization + +T0 = datetime(2020, 1, 1) +NAN = float("nan") + + +def _series(values): + return ([T0 + timedelta(hours=h) for h in range(len(values))], values) + + +class TestFitNormalization(unittest.TestCase): + def test_zscore_matches_numpy_across_chunks(self): + rng = np.random.default_rng(0) + data = rng.normal(5.0, 2.0, size=(100, 3)) + data[rng.random(data.shape) < 0.3] = np.nan + center, scale = fit_normalization(np.array_split(data, 7), "zscore") + np.testing.assert_allclose(center, np.nanmean(data, axis=0)) + np.testing.assert_allclose(scale, np.nanstd(data, axis=0)) + + def test_minmax(self): + data = np.array([[1.0, NAN], [3.0, 2.0], [NAN, 6.0]]) + center, scale = fit_normalization([data[:1], data[1:]], "minmax") + self.assertEqual(center, [1.0, 2.0]) + self.assertEqual(scale, [2.0, 4.0]) + + def test_constant_and_unobserved_features_get_unit_scale(self): + data = np.array([[4.0, NAN], [4.0, NAN]]) + center, scale = fit_normalization([data], "zscore") + self.assertEqual(center, [4.0, 0.0]) + self.assertEqual(scale, [1.0, 1.0]) + + def test_unknown_method(self): + with self.assertRaises(ValueError): + fit_normalization([np.zeros((1, 1))], "robust") + + +class TestTimeseriesProcessorOptions(unittest.TestCase): + def test_defaults_are_unchanged(self): + p = TimeseriesProcessor() + value = _series([[NAN, 1.0], [2.0, NAN]]) + p.fit([{"x": value}], "x") + self.assertFalse(p.learns_statistics) + self.assertEqual(p.size(), 2) + np.testing.assert_array_equal( + p.process(value).numpy(), [[0.0, 1.0], [2.0, 1.0]] + ) + + def test_zscore_normalization_with_indicators(self): + train = [{"x": _series([[1.0], [NAN], [3.0]])}, {"x": _series([[5.0]])}] + p = TimeseriesProcessor(normalization="zscore", missing_indicators=True) + p.fit(train, "x") + self.assertTrue(p.learns_statistics) + self.assertEqual(p.size(), 2) + mean, std = 3.0, np.std([1.0, 3.0, 5.0]) + + out = p.process(_series([[NAN], [5.0], [NAN]])).numpy() + # Leading gap -> 0 (the training mean), then forward-filled. + np.testing.assert_allclose(out[:, 0], [0.0, (5.0 - mean) / std, (5.0 - mean) / std]) + np.testing.assert_array_equal(out[:, 1], [1.0, 0.0, 1.0]) + + def test_invalid_normalization(self): + with self.assertRaises(ValueError): + TimeseriesProcessor(normalization="robust") + + +class TestTensorProcessorOptions(unittest.TestCase): + def test_defaults_are_unchanged(self): + p = TensorProcessor() + p.fit([{"x": [1.0, NAN]}], "x") + self.assertFalse(p.learns_statistics) + out = p.process([1.0, NAN]) + self.assertTrue(torch.isnan(out[1])) + + def test_normalization_indicators_and_zero_fill(self): + train = [{"x": [1.0, 10.0]}, {"x": [3.0, NAN]}] + p = TensorProcessor( + normalization="zscore", missing_indicators=True, impute_strategy="zero" + ) + p.fit(train, "x") + self.assertEqual(p.size(), 4) + out = p.process([NAN, 10.0]) + np.testing.assert_allclose(out.numpy(), [0.0, 0.0, 1.0, 0.0]) + out = p.process([3.0, NAN]) + np.testing.assert_allclose(out.numpy(), [1.0, 0.0, 0.0, 1.0]) + + +if __name__ == "__main__": + unittest.main() From 869ad6790886f7f6b3a349d7ca39adcb72f6d5fa Mon Sep 17 00:00:00 2001 From: Logiquo Date: Tue, 27 Oct 2026 03:05:22 -0400 Subject: [PATCH 2/4] Use built-in generics and add usage examples to the new processors --- pyhealth/processors/tensor_processor.py | 11 +++--- pyhealth/processors/timeseries_processor.py | 40 +++++++++++++++++---- pyhealth/processors/utils.py | 13 +++++-- 3 files changed, 49 insertions(+), 15 deletions(-) diff --git a/pyhealth/processors/tensor_processor.py b/pyhealth/processors/tensor_processor.py index 1bb30af8d..9e644ff22 100644 --- a/pyhealth/processors/tensor_processor.py +++ b/pyhealth/processors/tensor_processor.py @@ -1,4 +1,5 @@ -from typing import Any, Dict, Iterable, List, Literal, Optional +from typing import Any, Dict, Iterable, Optional +from typing import Literal import torch @@ -40,7 +41,7 @@ class TensorProcessor(FeatureProcessor): # Defaults for processors pickled before these options existed. normalization: Normalization = None missing_indicators: bool = False - impute_strategy: Optional[Literal["zero"]] = None + impute_strategy: Literal["zero"] | None = None def __init__( self, @@ -48,7 +49,7 @@ def __init__( spatial_dims: Optional[tuple[bool, ...]] = None, normalization: Normalization = None, missing_indicators: bool = False, - impute_strategy: Optional[Literal["zero"]] = None, + impute_strategy: Literal["zero"] | None = None, ): """ Initialize the TensorProcessor. @@ -77,8 +78,8 @@ def __init__( self.missing_indicators = missing_indicators self.impute_strategy = impute_strategy self.learns_statistics = normalization is not None - self.center: List[float] = [] - self.scale: List[float] = [] + self.center: list[float] = [] + self.scale: list[float] = [] def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None: """Infer n_dim and the feature width from the first valid sample. diff --git a/pyhealth/processors/timeseries_processor.py b/pyhealth/processors/timeseries_processor.py index c1335378a..b0f54d920 100644 --- a/pyhealth/processors/timeseries_processor.py +++ b/pyhealth/processors/timeseries_processor.py @@ -10,7 +10,7 @@ def sample_on_grid( - timestamps: List[datetime], values: np.ndarray, sampling_rate: timedelta + timestamps: list[datetime], values: np.ndarray, sampling_rate: timedelta ) -> np.ndarray: """Places values on a uniform time grid starting at the first timestamp. @@ -21,6 +21,15 @@ def sample_on_grid( Returns: Array of shape ``(S, F)``, NaN where the grid has no value. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors.timeseries_processor import sample_on_grid + >>> t0 = datetime(2020, 1, 1) + >>> sample_on_grid([t0, t0 + timedelta(hours=2)], [1.0, 3.0], timedelta(hours=1)) + array([[ 1.], + [nan], + [ 3.]]) """ values = np.asarray(values, dtype=float) if values.ndim == 1: @@ -105,8 +114,8 @@ def __init__( self.missing_indicators = missing_indicators self.learns_statistics = normalization is not None self.n_features = None - self.center: List[float] = [] - self.scale: List[float] = [] + self.center: list[float] = [] + self.scale: list[float] = [] def fit(self, samples: Any, field: str) -> None: """Fit the processor. @@ -141,7 +150,7 @@ def fit(self, samples: Any, field: str) -> None: ) def _sample( - self, timestamps: List[datetime], values: np.ndarray + self, timestamps: list[datetime], values: np.ndarray ) -> np.ndarray: return sample_on_grid(timestamps, values, self.sampling_rate) @@ -237,14 +246,22 @@ class TimeseriesLabelProcessor(FeatureProcessor): Args: sampling_rate: Interval of the sampled time steps. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor + >>> t0 = datetime(2020, 1, 1) + >>> times = [t0, t0 + timedelta(hours=2)] + >>> TimeseriesBinaryLabelProcessor().process((times, [0, 1])) + (tensor([0., 0., 1.]), tensor([1., 0., 1.])) """ def __init__(self, sampling_rate: timedelta = timedelta(hours=1)): self.sampling_rate = sampling_rate def process( - self, value: Tuple[List[datetime], np.ndarray] - ) -> Tuple[torch.Tensor, torch.Tensor]: + self, value: tuple[list[datetime], np.ndarray] + ) -> tuple[torch.Tensor, torch.Tensor]: timestamps, labels = value if len(timestamps) == 0: raise ValueError("Timestamps list is empty.") @@ -301,5 +318,14 @@ def _check(self, labels: np.ndarray) -> None: @register_processor("timeseries_regression") class TimeseriesRegressionLabelProcessor(TimeseriesLabelProcessor): - """Real-valued label per time step, see :class:`TimeseriesLabelProcessor`.""" + """Real-valued label per time step, see :class:`TimeseriesLabelProcessor`. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors import TimeseriesRegressionLabelProcessor + >>> t0 = datetime(2020, 1, 1) + >>> times = [t0 + timedelta(hours=h) for h in range(3)] + >>> TimeseriesRegressionLabelProcessor().process((times, [2.5, float("nan"), 0.5])) + (tensor([2.5000, 0.0000, 0.5000]), tensor([1., 0., 1.])) + """ diff --git a/pyhealth/processors/utils.py b/pyhealth/processors/utils.py index bd9d0bbb6..e61c8c3c5 100644 --- a/pyhealth/processors/utils.py +++ b/pyhealth/processors/utils.py @@ -1,16 +1,17 @@ """Utilities shared by processors.""" -from typing import Iterable, List, Literal, Optional, Tuple +from collections.abc import Iterable +from typing import Literal import numpy as np -Normalization = Optional[Literal["zscore", "minmax"]] +Normalization = Literal["zscore", "minmax"] | None NORMALIZATIONS = (None, "zscore", "minmax") def fit_normalization( arrays: Iterable[np.ndarray], method: Normalization -) -> Tuple[List[float], List[float]]: +) -> tuple[list[float], list[float]]: """Learns per-feature ``(center, scale)`` from the observed values. Missing values (``NaN``) are ignored. ``"zscore"`` uses the mean and the @@ -23,6 +24,12 @@ def fit_normalization( Returns: The center and scale of each feature, as lists of floats. + + Examples: + >>> import numpy as np + >>> from pyhealth.processors.utils import fit_normalization + >>> fit_normalization([np.array([[1.0], [3.0]]), np.array([[np.nan]])], "zscore") + ([2.0], [1.0]) """ count = total_mean = total_m2 = low = high = None for array in arrays: From 6f0548ca65e4b275220ddcb39c90c3481a63ac6e Mon Sep 17 00:00:00 2001 From: Logiquo Date: Tue, 27 Oct 2026 03:07:47 -0400 Subject: [PATCH 3/4] Sort the tensor processor imports --- pyhealth/processors/tensor_processor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pyhealth/processors/tensor_processor.py b/pyhealth/processors/tensor_processor.py index 9e644ff22..836f18411 100644 --- a/pyhealth/processors/tensor_processor.py +++ b/pyhealth/processors/tensor_processor.py @@ -1,5 +1,5 @@ -from typing import Any, Dict, Iterable, Optional -from typing import Literal +from collections.abc import Iterable +from typing import Any, Literal import torch @@ -46,7 +46,7 @@ class TensorProcessor(FeatureProcessor): def __init__( self, dtype: torch.dtype = torch.float32, - spatial_dims: Optional[tuple[bool, ...]] = None, + spatial_dims: tuple[bool, ...] | None = None, normalization: Normalization = None, missing_indicators: bool = False, impute_strategy: Literal["zero"] | None = None, @@ -81,7 +81,7 @@ def __init__( self.center: list[float] = [] self.scale: list[float] = [] - def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None: + def fit(self, samples: Iterable[dict[str, Any]], field: str) -> None: """Infer n_dim and the feature width from the first valid sample. Args: From 64b7ccac780093e4f298423eeb74c67f6f7279ab Mon Sep 17 00:00:00 2001 From: Logiquo Date: Tue, 27 Oct 2026 17:29:06 -0400 Subject: [PATCH 4/4] Share the time grid, reject label collisions and document the processors - Move sample_on_grid to processors/utils.py, next to a new grid_steps helper, and use it in TemporalTimeseriesProcessor too, so the gridding lives in one place. - TimeseriesLabelProcessor raises a ValueError when two labels fall into the same time step instead of silently keeping the last one, and its docstring now states how timestamps map to steps and that labels must share the input's first timestamp and sampling rate. - TensorProcessor: normalization keeps a scalar input scalar, and dim() / spatial() report one dimension when missing-value indicators turn a scalar into a vector. - Add API pages for the label processors and processors.utils, and note the new options in the processor list. - Replace the model-level tests that repeated test_timeseries_modes.py with one end-to-end test of the label processors; add tests for label collisions, flooring, the shared grid and scalar shapes. --- docs/api/processors.rst | 12 ++- ...cessors.TimeseriesBinaryLabelProcessor.rst | 9 ++ ...th.processors.TimeseriesLabelProcessor.rst | 9 ++ ...ors.TimeseriesRegressionLabelProcessor.rst | 9 ++ .../processors/pyhealth.processors.utils.rst | 8 ++ .../temporal_timeseries_processor.py | 39 ++++---- pyhealth/processors/tensor_processor.py | 18 ++-- pyhealth/processors/timeseries_processor.py | 71 ++++++--------- pyhealth/processors/utils.py | 64 +++++++++++++ tests/core/test_per_step_labels.py | 90 ++++++++----------- tests/core/test_processor_normalization.py | 40 ++++++++- 11 files changed, 236 insertions(+), 133 deletions(-) create mode 100644 docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst create mode 100644 docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst create mode 100644 docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst create mode 100644 docs/api/processors/pyhealth.processors.utils.rst diff --git a/docs/api/processors.rst b/docs/api/processors.rst index 0243890c7..f4f2996fb 100644 --- a/docs/api/processors.rst +++ b/docs/api/processors.rst @@ -35,6 +35,8 @@ Available Processors - ``MultiClassLabelProcessor``: Multi-class classification labels - ``MultiLabelProcessor``: Multi-label classification (e.g., drug recommendation) - ``RegressionLabelProcessor``: Continuous regression targets +- ``TimeseriesBinaryLabelProcessor``: Binary label per time step (``"timeseries_binary"``), returned as ``(value, mask)`` on the time grid of the input time series +- ``TimeseriesRegressionLabelProcessor``: Real-valued label per time step (``"timeseries_regression"``) **Data Type Processors:** @@ -43,9 +45,9 @@ Available Processors - ``TupleTimeTextProcessor``: For text paired with temporal information (time-aware text) - ``AudioProcessor``: For audio signal data - ``SignalProcessor``: For general signal data (e.g., EEG, ECG) -- ``TimeseriesProcessor``: For time-series data +- ``TimeseriesProcessor``: For time-series data; optional ``normalization`` (``"zscore"`` or ``"minmax"``, fitted on the training split) and ``missing_indicators`` - ``TimeImageProcessor``: For time-stamped image sequences (e.g., serial X-rays) -- ``TensorProcessor``: For pre-processed tensor data +- ``TensorProcessor``: For pre-processed tensor data; optional ``normalization``, ``missing_indicators`` and ``impute_strategy="zero"`` - ``RawProcessor``: Pass-through processor for raw data **Specialized Processors:** @@ -548,6 +550,9 @@ API Reference processors/pyhealth.processors.MultiClassLabelProcessor processors/pyhealth.processors.MultiLabelProcessor processors/pyhealth.processors.RegressionLabelProcessor + processors/pyhealth.processors.TimeseriesLabelProcessor + processors/pyhealth.processors.TimeseriesBinaryLabelProcessor + processors/pyhealth.processors.TimeseriesRegressionLabelProcessor processors/pyhealth.processors.ImageProcessor processors/pyhealth.processors.TextProcessor processors/pyhealth.processors.TupleTimeTextProcessor @@ -562,4 +567,5 @@ API Reference processors/pyhealth.processors.MultiHotProcessor processors/pyhealth.processors.StageNetProcessor processors/pyhealth.processors.StageNetTensorProcessor - processors/pyhealth.processors.GraphProcessor \ No newline at end of file + processors/pyhealth.processors.GraphProcessor + processors/pyhealth.processors.utils \ No newline at end of file diff --git a/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst new file mode 100644 index 000000000..378971fb3 --- /dev/null +++ b/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst @@ -0,0 +1,9 @@ +pyhealth.processors.TimeseriesBinaryLabelProcessor +================================================== + +Binary (0/1) label per time step (``"timeseries_binary"``). + +.. autoclass:: pyhealth.processors.TimeseriesBinaryLabelProcessor + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst new file mode 100644 index 000000000..009a0850c --- /dev/null +++ b/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst @@ -0,0 +1,9 @@ +pyhealth.processors.TimeseriesLabelProcessor +============================================ + +Base processor for labels given per time step, returning ``(value, mask)`` on the time grid of the input time series. + +.. autoclass:: pyhealth.processors.TimeseriesLabelProcessor + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst new file mode 100644 index 000000000..0aa0cd59a --- /dev/null +++ b/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst @@ -0,0 +1,9 @@ +pyhealth.processors.TimeseriesRegressionLabelProcessor +====================================================== + +Real-valued label per time step (``"timeseries_regression"``). + +.. autoclass:: pyhealth.processors.TimeseriesRegressionLabelProcessor + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/api/processors/pyhealth.processors.utils.rst b/docs/api/processors/pyhealth.processors.utils.rst new file mode 100644 index 000000000..8045fe3da --- /dev/null +++ b/docs/api/processors/pyhealth.processors.utils.rst @@ -0,0 +1,8 @@ +pyhealth.processors.utils +========================= + +Helpers shared by processors: learning normalization statistics and placing +values on a uniform time grid. + +.. automodule:: pyhealth.processors.utils + :members: diff --git a/pyhealth/processors/temporal_timeseries_processor.py b/pyhealth/processors/temporal_timeseries_processor.py index b38a84295..cbd3626c9 100644 --- a/pyhealth/processors/temporal_timeseries_processor.py +++ b/pyhealth/processors/temporal_timeseries_processor.py @@ -16,6 +16,7 @@ from . import register_processor from .base_processor import ModalityType, TemporalFeatureProcessor +from .utils import sample_on_grid @register_processor("temporal_timeseries") @@ -39,15 +40,18 @@ class TemporalTimeseriesProcessor(TemporalFeatureProcessor): sampling_rate: Uniform re-sampling interval. Defaults to 1 hour. impute_strategy: Currently only ``"forward_fill"`` is supported. - Example:: - - proc = TemporalTimeseriesProcessor(sampling_rate=timedelta(hours=2)) - from datetime import datetime, timedelta - ts = [datetime(2023,1,1,0), datetime(2023,1,1,4), datetime(2023,1,1,8)] - val = np.array([[120.0, 80.0], [115.0, 78.0], [118.0, 82.0]]) - out = proc.process_temporal((ts, val)) - # out["value"].shape → (5, 2) ← 5 two-hour steps over 8 h - # out["time"].shape → (5,) ← [0., 2., 4., 6., 8.] hours + Examples: + >>> from datetime import datetime, timedelta + >>> import numpy as np + >>> from pyhealth.processors import TemporalTimeseriesProcessor + >>> proc = TemporalTimeseriesProcessor(sampling_rate=timedelta(hours=2)) + >>> ts = [datetime(2023, 1, 1, 0), datetime(2023, 1, 1, 4), datetime(2023, 1, 1, 8)] + >>> val = np.array([[120.0, 80.0], [115.0, 78.0], [118.0, 82.0]]) + >>> out = proc.process_temporal((ts, val)) + >>> tuple(out["value"].shape) # 5 two-hour steps over 8 hours + (5, 2) + >>> out["time"].tolist() + [0.0, 2.0, 4.0, 6.0, 8.0] """ def __init__( @@ -89,21 +93,8 @@ def process(self, value: Tuple[List[datetime], np.ndarray]) -> dict: if len(timestamps) == 0: raise ValueError("Timestamps list is empty.") - values = np.asarray(values, dtype=float) - if values.ndim == 1: - values = values[:, None] # (T,) → (T, 1) - - num_features = values.shape[1] - start_time = timestamps[0] - end_time = timestamps[-1] - total_steps = int((end_time - start_time) / self.sampling_rate) + 1 - - sampled_values = np.full((total_steps, num_features), np.nan) - - for t, v in zip(timestamps, values): - idx = int((t - start_time) / self.sampling_rate) - if 0 <= idx < total_steps: - sampled_values[idx] = v + sampled_values = sample_on_grid(timestamps, values, self.sampling_rate) + total_steps, num_features = sampled_values.shape # Forward-fill imputation for f in range(num_features): diff --git a/pyhealth/processors/tensor_processor.py b/pyhealth/processors/tensor_processor.py index 836f18411..ad036029a 100644 --- a/pyhealth/processors/tensor_processor.py +++ b/pyhealth/processors/tensor_processor.py @@ -131,8 +131,10 @@ def process(self, value: Any) -> torch.Tensor: tensor = torch.tensor(value, dtype=self.dtype) if self.normalization is not None: - center = torch.tensor(self.center, dtype=self.dtype) - scale = torch.tensor(self.scale, dtype=self.dtype) + # Shaped like the last dimension, so a scalar input stays a scalar. + shape = tensor.shape[-1:] + center = torch.tensor(self.center, dtype=self.dtype).reshape(shape) + scale = torch.tensor(self.scale, dtype=self.dtype).reshape(shape) tensor = (tensor - center) / scale if self.missing_indicators: if tensor.dim() == 0: @@ -172,7 +174,8 @@ def dim(self) -> tuple[int, ...]: """Number of dimensions for the output tensor. Returns: - (n_dim,) + (n_dim,). With missing-value indicators, a scalar input becomes a + vector of 2, so n_dim is at least 1. Raises: NotImplementedError: If n_dim was not provided and fit() was not called. @@ -182,7 +185,12 @@ def dim(self) -> tuple[int, ...]: "TensorProcessor cannot determine n_dim automatically. " "Call fit() first." ) - return (self._n_dim,) + return (self._output_n_dim(),) + + def _output_n_dim(self) -> int: + if self.missing_indicators: + return max(self._n_dim, 1) + return self._n_dim def spatial(self) -> tuple[bool, ...]: """Whether each dimension of the output tensor is spatial. @@ -197,7 +205,7 @@ def spatial(self) -> tuple[bool, ...]: "TensorProcessor cannot determine spatial dims. " "Call fit() first." ) - return tuple(False for _ in range(self._n_dim)) + return tuple(False for _ in range(self._output_n_dim())) def __repr__(self) -> str: """ diff --git a/pyhealth/processors/timeseries_processor.py b/pyhealth/processors/timeseries_processor.py index b0f54d920..ef5a34ddd 100644 --- a/pyhealth/processors/timeseries_processor.py +++ b/pyhealth/processors/timeseries_processor.py @@ -6,47 +6,13 @@ from . import register_processor from .base_processor import FeatureProcessor -from .utils import NORMALIZATIONS, Normalization, fit_normalization - - -def sample_on_grid( - timestamps: list[datetime], values: np.ndarray, sampling_rate: timedelta -) -> np.ndarray: - """Places values on a uniform time grid starting at the first timestamp. - - Args: - timestamps: Time of each row of ``values``. - values: Array of shape ``(T,)`` or ``(T, F)``. - sampling_rate: Interval of the grid. - - Returns: - Array of shape ``(S, F)``, NaN where the grid has no value. - - Examples: - >>> from datetime import datetime, timedelta - >>> from pyhealth.processors.timeseries_processor import sample_on_grid - >>> t0 = datetime(2020, 1, 1) - >>> sample_on_grid([t0, t0 + timedelta(hours=2)], [1.0, 3.0], timedelta(hours=1)) - array([[ 1.], - [nan], - [ 3.]]) - """ - values = np.asarray(values, dtype=float) - if values.ndim == 1: - values = values[:, None] - num_features = values.shape[1] - - start_time = timestamps[0] - end_time = timestamps[-1] - total_steps = int((end_time - start_time) / sampling_rate) + 1 - sampled_values = np.full((total_steps, num_features), np.nan) - - # Map original timestamps to indices in the sampled grid - for t, v in zip(timestamps, values): - idx = int((t - start_time) / sampling_rate) - if 0 <= idx < total_steps: - sampled_values[idx] = v - return sampled_values +from .utils import ( + NORMALIZATIONS, + Normalization, + fit_normalization, + grid_steps, + sample_on_grid, +) @register_processor("timeseries") @@ -229,10 +195,16 @@ class TimeseriesLabelProcessor(FeatureProcessor): """Base processor for labels given per time step. The label is a time series ``(timestamps, values)``, placed on the same - kind of uniform grid as :class:`TimeseriesProcessor`. With the same first - timestamp and ``sampling_rate`` as the input time series, each label lines - up with one input step. Steps without a label (NaN or not on the grid) - are masked out and never imputed. + kind of uniform grid as :class:`TimeseriesProcessor`: a label at time + ``t`` goes to step ``floor((t - t0) / sampling_rate)``, where ``t0`` is the + first label timestamp. Steps that no label falls into, or whose label is + NaN, are masked out and never imputed. Two labels in the same step raise + a ``ValueError`` rather than one silently replacing the other. + + The processor does not see the input time series, so the caller must give + the labels the same first timestamp and ``sampling_rate`` as the input for + each label to line up with one input step. A label series longer than its + input is rejected by the model's loss. Output: - ``(value, mask)``: two float tensors of shape (S,), where ``mask`` @@ -265,6 +237,15 @@ def process( timestamps, labels = value if len(timestamps) == 0: raise ValueError("Timestamps list is empty.") + steps, counts = np.unique( + grid_steps(timestamps, self.sampling_rate), return_counts=True + ) + if (counts > 1).any(): + step = int(steps[counts > 1][0]) + raise ValueError( + f"{int(counts.max())} labels fall into time step {step}; " + "give at most one label per step of the sampling rate." + ) sampled = sample_on_grid(timestamps, labels, self.sampling_rate) if sampled.shape[1] != 1: raise ValueError("Expected one label per time step.") diff --git a/pyhealth/processors/utils.py b/pyhealth/processors/utils.py index e61c8c3c5..d1fc2fc2f 100644 --- a/pyhealth/processors/utils.py +++ b/pyhealth/processors/utils.py @@ -1,6 +1,7 @@ """Utilities shared by processors.""" from collections.abc import Iterable +from datetime import datetime, timedelta from typing import Literal import numpy as np @@ -75,3 +76,66 @@ def fit_normalization( raise ValueError(f"normalization must be one of {NORMALIZATIONS}") scale = np.where(scale > 0, scale, 1.0) return center.tolist(), scale.tolist() + + +def grid_steps(timestamps: list[datetime], sampling_rate: timedelta) -> np.ndarray: + """Step of each timestamp on a uniform grid starting at the first one. + + A timestamp ``t`` goes to step ``floor((t - t0) / sampling_rate)``. + + Args: + timestamps: Times, the first one starting the grid. + sampling_rate: Interval of the grid. + + Returns: + Integer array with the step of each timestamp. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors.utils import grid_steps + >>> t0 = datetime(2020, 1, 1) + >>> times = [t0, t0 + timedelta(minutes=90), t0 + timedelta(hours=3)] + >>> grid_steps(times, timedelta(hours=1)).tolist() + [0, 1, 3] + """ + start_time = timestamps[0] + return np.array( + [int((t - start_time) / sampling_rate) for t in timestamps], dtype=int + ) + + +def sample_on_grid( + timestamps: list[datetime], values: np.ndarray, sampling_rate: timedelta +) -> np.ndarray: + """Places values on a uniform time grid starting at the first timestamp. + + Each value goes to the step of its timestamp (see :func:`grid_steps`); + when several values fall into the same step, the last one is kept. + + Args: + timestamps: Time of each row of ``values``. + values: Array of shape ``(T,)`` or ``(T, F)``. + sampling_rate: Interval of the grid. + + Returns: + Array of shape ``(S, F)``, NaN where the grid has no value. + + Examples: + >>> from datetime import datetime, timedelta + >>> from pyhealth.processors.utils import sample_on_grid + >>> t0 = datetime(2020, 1, 1) + >>> sample_on_grid([t0, t0 + timedelta(hours=2)], [1.0, 3.0], timedelta(hours=1)) + array([[ 1.], + [nan], + [ 3.]]) + """ + values = np.asarray(values, dtype=float) + if values.ndim == 1: + values = values[:, None] + steps = grid_steps(timestamps, sampling_rate) + total_steps = int((timestamps[-1] - timestamps[0]) / sampling_rate) + 1 + sampled_values = np.full((total_steps, values.shape[1]), np.nan) + for step, value in zip(steps, values): + if 0 <= step < total_steps: + sampled_values[step] = value + return sampled_values diff --git a/tests/core/test_per_step_labels.py b/tests/core/test_per_step_labels.py index bfeb3f65d..4268be3b0 100644 --- a/tests/core/test_per_step_labels.py +++ b/tests/core/test_per_step_labels.py @@ -62,61 +62,43 @@ def test_binary_rejects_other_values(self): with self.assertRaises(ValueError): TimeseriesBinaryLabelProcessor().process((_times(2), [0, 2])) - -class TestPerStepModel(unittest.TestCase): - def test_modes_are_separate_from_binary_and_regression(self): - model = RNN(dataset=_dataset("timeseries_binary")) - self.assertEqual(model.mode, "timeseries_binary") - model = RNN(dataset=_dataset("timeseries_regression")) - self.assertEqual(model.mode, "timeseries_regression") - - def test_loss_ignores_unlabeled_and_padded_steps(self): - # Stay 1 has an unlabeled step and is padded; changing what the - # model predicts there must not change the loss. - dataset = _dataset(label_fn=lambda i, h: NAN if (i, h) == (1, 0) else float(h % 2)) - model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval() - batch = next(iter(get_dataloader(dataset, batch_size=4))) - with torch.no_grad(): + def test_timestamps_between_steps_are_floored(self): + times = [T0, T0 + timedelta(minutes=90)] + value, mask = TimeseriesRegressionLabelProcessor().process((times, [1.0, 2.0])) + np.testing.assert_array_equal(value.numpy(), [1.0, 2.0]) + np.testing.assert_array_equal(mask.numpy(), [1, 1]) + + def test_two_labels_in_one_step_are_rejected(self): + times = [T0, T0 + timedelta(minutes=20), T0 + timedelta(hours=1)] + with self.assertRaisesRegex(ValueError, "time step 0"): + TimeseriesBinaryLabelProcessor().process((times, [0, 1, 0])) + + +class TestPerStepEndToEnd(unittest.TestCase): + """The label processors with a model and the Trainer. + + Model-level behaviour (masked loss, causality, bidirectional guard) is + covered with a mock label processor in test_timeseries_modes.py. + """ + + def test_label_processors_train_and_evaluate(self): + for mode, metric in ( + ("timeseries_binary", "roc_auc"), + ("timeseries_regression", "mae"), + ): + dataset = _dataset(mode) + model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8) + self.assertEqual(model.mode, mode) + batch = next(iter(get_dataloader(dataset, batch_size=4))) out = model(**batch) - value, mask = out["y_true"] - self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1)) - self.assertEqual(int(mask.sum()), 17) # 5 + 3 + 6 + 4 steps, minus one - - logits = out["logit"].clone() - loss_fn = model.get_loss_function() - logits[mask == 0] = 1e3 - torch.testing.assert_close(loss_fn(logits, (value, mask)), out["loss"]) - - def test_predictions_do_not_see_future_steps(self): - dataset = _dataset() - model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval() - batch = next(iter(get_dataloader(dataset, batch_size=1))) - with torch.no_grad(): - before = model(**batch)["y_prob"][0, :, 0] - batch["dynamic"] = batch["dynamic"].clone() - batch["dynamic"][:, -1] += 100.0 - after = model(**batch)["y_prob"][0, :, 0] - torch.testing.assert_close(before[:-1], after[:-1]) - self.assertFalse(torch.allclose(before[-1], after[-1])) - - def test_bidirectional_is_rejected(self): - with self.assertRaises(ValueError): - RNN(dataset=_dataset(), bidirectional=True) - - def test_trainer_scores_labeled_steps_per_sample(self): - dataset = _dataset() - model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8) - trainer = Trainer(model=model, metrics=["roc_auc"], enable_logging=False) - loader = get_dataloader(dataset, batch_size=3) - y_true, y_prob, _ = trainer.inference(loader) - self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6]) - self.assertEqual([len(p) for p in y_prob], [len(y) for y in y_true]) - self.assertIn("roc_auc", trainer.evaluate(loader)) - - dataset = _dataset("timeseries_regression") - model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8) - trainer = Trainer(model=model, metrics=["mae"], enable_logging=False) - self.assertIn("mae", trainer.evaluate(get_dataloader(dataset, batch_size=3))) + self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1)) + self.assertTrue(torch.isfinite(out["loss"])) + + trainer = Trainer(model=model, metrics=[metric], enable_logging=False) + loader = get_dataloader(dataset, batch_size=3) + y_true, _, _ = trainer.inference(loader) + self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6]) + self.assertIn(metric, trainer.evaluate(loader)) if __name__ == "__main__": unittest.main() diff --git a/tests/core/test_processor_normalization.py b/tests/core/test_processor_normalization.py index 99dec0a70..b546e51d0 100644 --- a/tests/core/test_processor_normalization.py +++ b/tests/core/test_processor_normalization.py @@ -4,8 +4,12 @@ import numpy as np import torch -from pyhealth.processors import TensorProcessor, TimeseriesProcessor -from pyhealth.processors.utils import fit_normalization +from pyhealth.processors import ( + TemporalTimeseriesProcessor, + TensorProcessor, + TimeseriesProcessor, +) +from pyhealth.processors.utils import fit_normalization, grid_steps, sample_on_grid T0 = datetime(2020, 1, 1) NAN = float("nan") @@ -41,6 +45,25 @@ def test_unknown_method(self): fit_normalization([np.zeros((1, 1))], "robust") +class TestGrid(unittest.TestCase): + def test_grid_steps_floor_timestamps(self): + times = [T0, T0 + timedelta(minutes=59), T0 + timedelta(minutes=61)] + self.assertEqual(grid_steps(times, timedelta(hours=1)).tolist(), [0, 0, 1]) + + def test_sample_on_grid_keeps_the_last_value_of_a_step(self): + times = [T0, T0 + timedelta(minutes=30), T0 + timedelta(hours=2)] + out = sample_on_grid(times, [1.0, 2.0, 3.0], timedelta(hours=1)) + np.testing.assert_array_equal(out[:, 0], [2.0, NAN, 3.0]) + + def test_temporal_timeseries_processor_uses_the_shared_grid(self): + value = ([T0, T0 + timedelta(hours=2)], np.array([[1.0], [3.0]])) + p = TemporalTimeseriesProcessor() + p.fit([{"x": value}], "x") + out = p.process(value) + np.testing.assert_array_equal(out["value"].numpy()[:, 0], [1.0, 1.0, 3.0]) + np.testing.assert_array_equal(out["time"].numpy(), [0.0, 1.0, 2.0]) + + class TestTimeseriesProcessorOptions(unittest.TestCase): def test_defaults_are_unchanged(self): p = TimeseriesProcessor() @@ -90,6 +113,19 @@ def test_normalization_indicators_and_zero_fill(self): out = p.process([3.0, NAN]) np.testing.assert_allclose(out.numpy(), [1.0, 0.0, 0.0, 1.0]) + def test_scalar_shapes_match_dim(self): + for kwargs, shape in ( + ({"normalization": "zscore"}, ()), + ({"missing_indicators": True}, (2,)), + ({"normalization": "zscore", "missing_indicators": True}, (2,)), + ): + p = TensorProcessor(**kwargs) + p.fit([{"x": 1.0}, {"x": 3.0}], "x") + out = p.process(2.0) + self.assertEqual(tuple(out.shape), shape) + self.assertEqual(p.dim(), (len(shape),)) + self.assertEqual(len(p.spatial()), len(shape)) + if __name__ == "__main__": unittest.main()