8969779898

Committed 06 May 2024 01:16PM UTC coverage: 81.801% (+0.1%) from 81.682%

Build # 8969779898

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #387 from WenjieDu/dev

Refactor code, update docstring, and add README in Chinese

Run Details

20 of 52 new or added lines in 32 files covered. (38.46%)

10 existing lines in 6 files now uncovered.

7776 of 9506 relevant lines covered (81.8%)

4.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.24

/pypots/imputation/locf/model.py

"""
The implementation of LOCF (Last Observed Carried Forward) for the partially-observed time-series imputation task.

"""

# Created by Wenjie Du <wenjay.du@gmail.com>
# License: BSD-3-Clause

import warnings
from typing import Union, Optional

import h5py
import numpy as np
import torch

from .core import locf_numpy, locf_torch
from ..base import BaseImputer


class LOCF(BaseImputer):
    """LOCF (Last Observed Carried Forward) imputation method. A naive imputation method that fills missing values
    with the last observed value. When time-series data gets inverse on the time dimension, this method can also be
    seen as NOCB (Next Observation Carried Backward). Simple but commonly used in practice.

    Parameters
    ----------
    first_step_imputation : str, default='backward'
        With LOCF, the observed values are carried forward to impute the missing ones. But if the first value
        is missing, there is no value to carry forward. This parameter is used to determine the strategy to
        impute the missing values at the beginning of the time-series sequence after LOCF is applied.
        It can be one of ['backward', 'zero', 'median', 'nan'].
        If 'nan', the missing values at the sequence beginning will be left as NaNs.
        If 'zero', the missing values at the sequence beginning will be imputed with 0.
        If 'backward', the missing values at the beginning of the time-series sequence will be imputed with the
        first observed value in the sequence, i.e. the first observed value will be carried backward to impute
        the missing values at the beginning of the sequence. This method is also known as NOCB (Next Observation
        Carried Backward). If 'median', the missing values at the sequence beginning will be imputed with the overall
        median values of features in the dataset.
        If `first_step_imputation` is not "nan", if missing values still exist (this is usually caused by whole feature
        missing) after applying `first_step_imputation`, they will be filled with 0.

    """

    def __init__(
        self,
        first_step_imputation: str = "zero",
        device: Optional[Union[str, torch.device, list]] = None,
    ):
        super().__init__(device=device)
        assert first_step_imputation in ["nan", "zero", "backward", "median"]
        self.first_step_imputation = first_step_imputation

    def fit(
        self,
        train_set: Union[dict, str],
        val_set: Optional[Union[dict, str]] = None,
        file_type: str = "hdf5",
    ) -> None:
        """Train the imputer on the given data.

        Warnings
        --------
        LOCF does not need to run fit().
        Please run func ``predict()`` directly.

        """
        warnings.warn(
            "LOCF (Last Observed Carried Forward) imputation class has no parameter to train. "
            "Please run func `predict()` directly."
        )

    def predict(
        self,
        test_set: Union[dict, str],
        file_type: str = "hdf5",
    ) -> dict:
        """Make predictions for the input data with the trained model.

        Parameters
        ----------
        test_set : dict or str
            The dataset for model validating, should be a dictionary including keys as 'X',
            or a path string locating a data file supported by PyPOTS (e.g. h5 file).
            If it is a dict, X should be array-like of shape [n_samples, sequence length (n_steps), n_features],
            which is time-series data for validating, can contain missing values, and y should be array-like of shape
            [n_samples], which is classification labels of X.
            If it is a path string, the path should point to a data file, e.g. a h5 file, which contains
            key-value pairs like a dict, and it has to include keys as 'X' and 'y'.

        file_type :
            The type of the given file if test_set is a path string.

        Returns
        -------
        result_dict: dict
            Prediction results in a Python Dictionary for the given samples.
            It should be a dictionary including keys as 'imputation', 'classification', 'clustering', and 'forecasting'.
            For sure, only the keys that relevant tasks are supported by the model will be returned.
        """
        if isinstance(test_set, str):
            with h5py.File(test_set, "r") as f:
                X = f["X"][:]
        else:
            X = test_set["X"]

        assert len(X.shape) == 3, (
            f"Input X should have 3 dimensions [n_samples, n_steps, n_features], "
            f"but the actual shape of X: {X.shape}"
        )
        if isinstance(X, list):
            X = np.asarray(X)

        if isinstance(X, np.ndarray):
            imputed_data = locf_numpy(X, self.first_step_imputation)
        elif isinstance(X, torch.Tensor):
            imputed_data = locf_torch(X, self.first_step_imputation)
        else:
            raise TypeError(
                "X must be type of list/np.ndarray/torch.Tensor, " f"but got {type(X)}"
            )

        result_dict = {
            "imputation": imputed_data,
        }
        return result_dict

    def impute(
        self,
        test_set: Union[dict, str],
        file_type: str = "hdf5",
    ) -> np.ndarray:
        """Impute missing values in the given data with the trained model.

        Parameters
        ----------
        test_set :
            The data samples for testing, should be array-like of shape [n_samples, sequence length (n_steps),
            n_features], or a path string locating a data file, e.g. h5 file.

        file_type :
            The type of the given file if X is a path string.

        Returns
        -------
        array-like, shape [n_samples, sequence length (n_steps), n_features],
            Imputed data.
        """

        result_dict = self.predict(test_set, file_type=file_type)
        return result_dict["imputation"]

1	"""	6✔
2	The implementation of LOCF (Last Observed Carried Forward) for the partially-observed time-series imputation task.
3
4	"""
5
6	# Created by Wenjie Du <wenjay.du@gmail.com>
7	# License: BSD-3-Clause
8
9	import warnings	6✔
10	from typing import Union, Optional	6✔
11
12	import h5py	6✔
13	import numpy as np	6✔
14	import torch	6✔
15
16	from .core import locf_numpy, locf_torch	6✔
17	from ..base import BaseImputer	6✔
18
19
20	class LOCF(BaseImputer):	6✔
21	"""LOCF (Last Observed Carried Forward) imputation method. A naive imputation method that fills missing values	6✔
22	with the last observed value. When time-series data gets inverse on the time dimension, this method can also be
23	seen as NOCB (Next Observation Carried Backward). Simple but commonly used in practice.
24
25	Parameters
26	----------
27	first_step_imputation : str, default='backward'
28	With LOCF, the observed values are carried forward to impute the missing ones. But if the first value
29	is missing, there is no value to carry forward. This parameter is used to determine the strategy to
30	impute the missing values at the beginning of the time-series sequence after LOCF is applied.
31	It can be one of ['backward', 'zero', 'median', 'nan'].
32	If 'nan', the missing values at the sequence beginning will be left as NaNs.
33	If 'zero', the missing values at the sequence beginning will be imputed with 0.
34	If 'backward', the missing values at the beginning of the time-series sequence will be imputed with the
35	first observed value in the sequence, i.e. the first observed value will be carried backward to impute
36	the missing values at the beginning of the sequence. This method is also known as NOCB (Next Observation
37	Carried Backward). If 'median', the missing values at the sequence beginning will be imputed with the overall
38	median values of features in the dataset.
39	If `first_step_imputation` is not "nan", if missing values still exist (this is usually caused by whole feature
40	missing) after applying `first_step_imputation`, they will be filled with 0.
41
42	"""
43
44	def __init__(	6✔
45	self,
46	first_step_imputation: str = "zero",
47	device: Optional[Union[str, torch.device, list]] = None,
48	):
49	super().__init__(device=device)	6✔
50	assert first_step_imputation in ["nan", "zero", "backward", "median"]	6✔
51	self.first_step_imputation = first_step_imputation	6✔
52
53	def fit(	6✔
54	self,
55	train_set: Union[dict, str],
56	val_set: Optional[Union[dict, str]] = None,
57	file_type: str = "hdf5",
58	) -> None:
59	"""Train the imputer on the given data.
60
61	Warnings
62	--------
63	LOCF does not need to run fit().
64	Please run func ``predict()`` directly.
65
66	"""
67	warnings.warn(	6✔
68	"LOCF (Last Observed Carried Forward) imputation class has no parameter to train. "
69	"Please run func `predict()` directly."
70	)
71
72	def predict(	6✔
73	self,
74	test_set: Union[dict, str],
75	file_type: str = "hdf5",
76	) -> dict:
77	"""Make predictions for the input data with the trained model.
78
79	Parameters
80	----------
81	test_set : dict or str
82	The dataset for model validating, should be a dictionary including keys as 'X',
83	or a path string locating a data file supported by PyPOTS (e.g. h5 file).
84	If it is a dict, X should be array-like of shape [n_samples, sequence length (n_steps), n_features],
85	which is time-series data for validating, can contain missing values, and y should be array-like of shape
86	[n_samples], which is classification labels of X.
87	If it is a path string, the path should point to a data file, e.g. a h5 file, which contains
88	key-value pairs like a dict, and it has to include keys as 'X' and 'y'.
89
90	file_type :
91	The type of the given file if test_set is a path string.
92
93	Returns
94	-------
95	result_dict: dict
96	Prediction results in a Python Dictionary for the given samples.
97	It should be a dictionary including keys as 'imputation', 'classification', 'clustering', and 'forecasting'.
98	For sure, only the keys that relevant tasks are supported by the model will be returned.
99	"""
100	if isinstance(test_set, str):	6✔
101	with h5py.File(test_set, "r") as f:	6✔
102	X = f["X"][:]	6✔
103	else:
104	X = test_set["X"]	6✔
105
106	assert len(X.shape) == 3, (	6✔
107	f"Input X should have 3 dimensions [n_samples, n_steps, n_features], "
108	f"but the actual shape of X: {X.shape}"
109	)
110	if isinstance(X, list):	6✔
111	X = np.asarray(X)	×
112
113	if isinstance(X, np.ndarray):	6✔
114	imputed_data = locf_numpy(X, self.first_step_imputation)	6✔
115	elif isinstance(X, torch.Tensor):	6✔
116	imputed_data = locf_torch(X, self.first_step_imputation)	6✔
117	else:
118	raise TypeError(	×
119	"X must be type of list/np.ndarray/torch.Tensor, " f"but got {type(X)}"
120	)
121
122	result_dict = {	6✔
123	"imputation": imputed_data,
124	}
125	return result_dict	6✔
126
127	def impute(	6✔
128	self,
129	test_set: Union[dict, str],
130	file_type: str = "hdf5",
131	) -> np.ndarray:
132	"""Impute missing values in the given data with the trained model.
133
134	Parameters
135	----------
136	test_set :
137	The data samples for testing, should be array-like of shape [n_samples, sequence length (n_steps),
138	n_features], or a path string locating a data file, e.g. h5 file.
139
140	file_type :
141	The type of the given file if X is a path string.
142
143	Returns
144	-------
145	array-like, shape [n_samples, sequence length (n_steps), n_features],
146	Imputed data.
147	"""
148
NEW 149	result_dict = self.predict(test_set, file_type=file_type)	×
NEW 150	return result_dict["imputation"]	×

WenjieDu / PyPOTS / 8969779898

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous