arundo / adtk / 203

pending completion

Build # 203

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/arundo/adtk/commit/<a class=hub.com/arundo/adtk/commit/12cb8b1b66c2fcf37b6d58fab06d93c5794a29f5">12cb8b1b6<a href="https://github.com/arundo/adtk/commit/12cb8b1b66c2fcf37b6d58fab06d93c5794a29f5">&quot;&gt;Revert &amp;quot;Type hints (#79)&amp;quot; (#84)

This reverts commit </a><a class="double-link" href="https://github.com/arundo/adtk/commit/<a class="double-link" href="https://github.com/arundo/adtk/commit/b7b229688a3f52333576945d64386ca63f49cbf0">b7b229688</a>">b7b229688</a><a href="https://github.com/arundo/adtk/commit/12cb8b1b66c2fcf37b6d58fab06d93c5794a29f5">.

Run Details

496 of 496 new or added lines in 20 files covered. (100.0%)

1480 of 1510 relevant lines covered (98.01%)

3.92 hits per line

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

99.0

/src/adtk/_base.py

from abc import ABC, abstractmethod
from copy import deepcopy

import pandas as pd


class _Model(ABC):
    _need_fit = True
    _default_params = {}

    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)
        self._fitted = (
            0
        )  # 0 for not fitted, 1 for fitted, 2 for univariate model fitted by DF

    @abstractmethod
    def _fit(self, ts):
        pass

    @abstractmethod
    def _predict(self, ts):
        pass

    def get_params(self):
        """Get parameters of this model.

        Returns
        -------
        dict
            Model parameters.

        """
        return {key: getattr(self, key) for key in self._default_params.keys()}

    def set_params(self, **kwargs):
        """Set parameters of this model.

        Parameters
        ----------
        **kwargs
            Model parameters to set. If empty, then all parameters will be
            reset to default values.

        """
        for key, value in kwargs.items():
            if key not in self._default_params.keys():
                raise KeyError("{} is not a valid parameter".format(key))
        for key, value in kwargs.items():
            setattr(self, key, value)

        if not kwargs:
            for key, value in self._default_params.items():
                setattr(self, key, value)


class _Model1D(_Model):
    """Base class of _Detector1D and _Transformer1D."""

    def __init__(self, **kwargs):
        self._models = None
        super().__init__(**kwargs)

    def _update_models(self, cols):
        """Update attribute _models with given columns and model parameters.

        When a 1D model applied to a DataFrame, it is applied independently to
        each column of the DataFrame. Internally, a model object is created for
        each column. This private method is used to set those models.

        """
        # initialize or reset a model for each column
        if self._models is None:
            self._models = dict()
        for col in cols:
            if col not in self._models:
                self._models[col] = self.__class__()
            else:
                self._models[col].set_params()
        # remove models that correspond no column
        models_to_remove = list(
            filter(lambda key: key not in cols, self._models.keys())
        )
        for key in models_to_remove:
            self._models.pop(key)
        # for each parameter, set them over all models.
        for col in cols:
            self._models[col].set_params(**deepcopy(self.get_params()))

    def _fit(self, ts):
        if isinstance(ts, pd.Series):
            s = ts.copy()
            self._fit_core(s)
            self._models = None
            self._fitted = 1
        elif isinstance(ts, pd.DataFrame):
            df = ts.copy()
            if df.columns.duplicated().any():
                raise ValueError(
                    "Input DataFrame must have unique column names."
                )
            if self._need_fit:
                self._update_models(df.columns)
                # fit model for each column
                for col in df.columns:
                    self._models[col].fit(df[col])
                self._fitted = 2
            else:
                pass
        else:
            raise TypeError("Input must be a pandas Series or DataFrame.")

    def _predict(self, ts):
        if self._need_fit and (self._fitted == 0):
            raise RuntimeError("The model must be trained first.")
        if isinstance(ts, pd.Series):
            if self._need_fit and (
                self._fitted == 2
            ):  # fitted by DF, to be applied to Series
                raise RuntimeError(
                    "The model was trained by a pandas DataFrame object, "
                    "it can only be applied to a pandas DataFrame object."
                )
            s = ts.copy()
            predicted = self._predict_core(s)
            # if a Series-to-Series operation, make sure Series name keeps
            if isinstance(predicted, pd.Series):
                predicted.name = ts.name
        elif isinstance(ts, pd.DataFrame):
            df = ts.copy()
            if df.columns.duplicated().any():
                raise ValueError(
                    "Input DataFrame must have unique column names."
                )
            if (not self._need_fit) or (self._fitted == 1):
                # apply the model to each column
                predicted = []
                for col in df.columns:
                    predicted_this_col = self._predict(df[col])
                    if isinstance(predicted_this_col, pd.DataFrame):
                        predicted_this_col = predicted_this_col.rename(
                            columns={
                                col1: "{}_{}".format(col, col1)
                                for col1 in predicted_this_col.columns
                            }
                        )
                    predicted.append(predicted_this_col)
                predicted = pd.concat(predicted, axis=1)
            else:
                # predict for each column
                if not (set(self._models.keys()) >= set(df.columns)):
                    raise ValueError(
                        "The model was trained by a pandas DataFrame with "
                        "columns {}, but the input DataFrame contains columns "
                        "{} which are unknown to the model.".format(
                            list(set(self._models.keys())),
                            list(set(df.columns) - set(self._models.keys())),
                        )
                    )
                predicted = pd.concat(
                    [
                        self._models[col]._predict(df[col])
                        for col in df.columns
                    ],
                    axis=1,
                )
        else:
            raise TypeError("Input must be a pandas Series or DataFrame.")
        # make sure index freq is the same (because pandas has a bug that some
        # operation, e.g. concat, may change freq)
        predicted.index.freq = ts.index.freq
        return predicted

    @abstractmethod
    def _fit_core(self, ts):
        pass

    @abstractmethod
    def _predict_core(self, ts):
        pass

    @abstractmethod
    def fit(self, ts):
        pass

    @abstractmethod
    def predict(self, ts):
        pass

    @abstractmethod
    def fit_predict(self, ts):
        pass


class _ModelHD(_Model):
    def _fit(self, df):
        if isinstance(df, pd.DataFrame):
            if df.columns.duplicated().any():
                raise ValueError(
                    "Input DataFrame must have unique column names."
                )
            df_copy = df.copy()
            self._fit_core(df_copy)
        else:
            raise TypeError("Input must be a pandas DataFrame.")
        self._fitted = 1

    def _predict(self, df):
        if self._need_fit and (self._fitted == 0):
            raise RuntimeError("The model must be trained first.")
        if isinstance(df, pd.DataFrame):
            if df.columns.duplicated().any():
                raise ValueError(
                    "Input DataFrame must have unique column names."
                )
            df_copy = df.copy()
            predicted = self._predict_core(df_copy)
        else:
            raise TypeError("Input must be a pandas DataFrame.")
        # make sure index freq is the same (because pandas has a bug that some
        # operation, e.g. concat, may change freq)
        predicted.index.freq = df.index.freq
        return predicted

    @abstractmethod
    def _fit_core(self, df):
        pass

    @abstractmethod
    def _predict_core(self, df):
        pass

    @abstractmethod
    def fit(self, df):
        pass

    @abstractmethod
    def predict(self, df):
        pass

    @abstractmethod
    def fit_predict(self, df):
        pass

1	from abc import ABC, abstractmethod	4✔
2	from copy import deepcopy	4✔
3
4	import pandas as pd	4✔
5
6
7	class _Model(ABC):	4✔
8	_need_fit = True	4✔
9	_default_params = {}	4✔
10
11	def __init__(self, **kwargs):	4✔
12	for key, value in kwargs.items():	4✔
13	setattr(self, key, value)	4✔
14	self._fitted = (	4✔
15	0
16	) # 0 for not fitted, 1 for fitted, 2 for univariate model fitted by DF
17
18	@abstractmethod	4✔
19	def _fit(self, ts):	1✔
20	pass
21
22	@abstractmethod	4✔
23	def _predict(self, ts):	1✔
24	pass
25
26	def get_params(self):	4✔
27	"""Get parameters of this model.
28
29	Returns
30	-------
31	dict
32	Model parameters.
33
34	"""
35	return {key: getattr(self, key) for key in self._default_params.keys()}	4✔
36
37	def set_params(self, **kwargs):	4✔
38	"""Set parameters of this model.
39
40	Parameters
41	----------
42	**kwargs
43	Model parameters to set. If empty, then all parameters will be
44	reset to default values.
45
46	"""
47	for key, value in kwargs.items():	4✔
48	if key not in self._default_params.keys():	4✔
49	raise KeyError("{} is not a valid parameter".format(key))
50	for key, value in kwargs.items():	4✔
51	setattr(self, key, value)	4✔
52
53	if not kwargs:	4✔
54	for key, value in self._default_params.items():	4✔
55	setattr(self, key, value)	4✔
56
57
58	class _Model1D(_Model):	4✔
59	"""Base class of _Detector1D and _Transformer1D."""
60
61	def __init__(self, **kwargs):	4✔
62	self._models = None	4✔
63	super().__init__(**kwargs)	4✔
64
65	def _update_models(self, cols):	4✔
66	"""Update attribute _models with given columns and model parameters.
67
68	When a 1D model applied to a DataFrame, it is applied independently to
69	each column of the DataFrame. Internally, a model object is created for
70	each column. This private method is used to set those models.
71
72	"""
73	# initialize or reset a model for each column
74	if self._models is None:	4✔
75	self._models = dict()	4✔
76	for col in cols:	4✔
77	if col not in self._models:	4✔
78	self._models[col] = self.__class__()	4✔
79	else:
80	self._models[col].set_params()	4✔
81	# remove models that correspond no column
82	models_to_remove = list(	4✔
83	filter(lambda key: key not in cols, self._models.keys())
84	)
85	for key in models_to_remove:	4✔
86	self._models.pop(key)	×
87	# for each parameter, set them over all models.
88	for col in cols:	4✔
89	self._models[col].set_params(**deepcopy(self.get_params()))	4✔
90
91	def _fit(self, ts):	4✔
92	if isinstance(ts, pd.Series):	4✔
93	s = ts.copy()	4✔
94	self._fit_core(s)	4✔
95	self._models = None	4✔
96	self._fitted = 1	4✔
97	elif isinstance(ts, pd.DataFrame):	4✔
98	df = ts.copy()	4✔
99	if df.columns.duplicated().any():	4✔
100	raise ValueError(
101	"Input DataFrame must have unique column names."
102	)
103	if self._need_fit:	4✔
104	self._update_models(df.columns)	4✔
105	# fit model for each column
106	for col in df.columns:	4✔
107	self._models[col].fit(df[col])	4✔
108	self._fitted = 2	4✔
109	else:
110	pass
111	else:
112	raise TypeError("Input must be a pandas Series or DataFrame.")
113
114	def _predict(self, ts):	4✔
115	if self._need_fit and (self._fitted == 0):	4✔
116	raise RuntimeError("The model must be trained first.")
117	if isinstance(ts, pd.Series):	4✔
118	if self._need_fit and (	4✔
119	self._fitted == 2
120	): # fitted by DF, to be applied to Series
121	raise RuntimeError(
122	"The model was trained by a pandas DataFrame object, "
123	"it can only be applied to a pandas DataFrame object."
124	)
125	s = ts.copy()	4✔
126	predicted = self._predict_core(s)	4✔
127	# if a Series-to-Series operation, make sure Series name keeps
128	if isinstance(predicted, pd.Series):	4✔
129	predicted.name = ts.name	4✔
130	elif isinstance(ts, pd.DataFrame):	4✔
131	df = ts.copy()	4✔
132	if df.columns.duplicated().any():	4✔
133	raise ValueError(
134	"Input DataFrame must have unique column names."
135	)
136	if (not self._need_fit) or (self._fitted == 1):	4✔
137	# apply the model to each column
138	predicted = []	4✔
139	for col in df.columns:	4✔
140	predicted_this_col = self._predict(df[col])	4✔
141	if isinstance(predicted_this_col, pd.DataFrame):	4✔
142	predicted_this_col = predicted_this_col.rename(	4✔
143	columns={
144	col1: "{}_{}".format(col, col1)
145	for col1 in predicted_this_col.columns
146	}
147	)
148	predicted.append(predicted_this_col)	4✔
149	predicted = pd.concat(predicted, axis=1)	4✔
150	else:
151	# predict for each column
152	if not (set(self._models.keys()) >= set(df.columns)):	4✔
153	raise ValueError(
154	"The model was trained by a pandas DataFrame with "
155	"columns {}, but the input DataFrame contains columns "
156	"{} which are unknown to the model.".format(
157	list(set(self._models.keys())),
158	list(set(df.columns) - set(self._models.keys())),
159	)
160	)
161	predicted = pd.concat(	4✔
162	[
163	self._models[col]._predict(df[col])
164	for col in df.columns
165	],
166	axis=1,
167	)
168	else:
169	raise TypeError("Input must be a pandas Series or DataFrame.")
170	# make sure index freq is the same (because pandas has a bug that some
171	# operation, e.g. concat, may change freq)
172	predicted.index.freq = ts.index.freq	4✔
173	return predicted	4✔
174
175	@abstractmethod	4✔
176	def _fit_core(self, ts):	1✔
177	pass
178
179	@abstractmethod	4✔
180	def _predict_core(self, ts):	1✔
181	pass
182
183	@abstractmethod	4✔
184	def fit(self, ts):	1✔
185	pass
186
187	@abstractmethod	4✔
188	def predict(self, ts):	1✔
189	pass
190
191	@abstractmethod	4✔
192	def fit_predict(self, ts):	1✔
193	pass
194
195
196	class _ModelHD(_Model):	4✔
197	def _fit(self, df):	4✔
198	if isinstance(df, pd.DataFrame):	4✔
199	if df.columns.duplicated().any():	4✔
200	raise ValueError(
201	"Input DataFrame must have unique column names."
202	)
203	df_copy = df.copy()	4✔
204	self._fit_core(df_copy)	4✔
205	else:
206	raise TypeError("Input must be a pandas DataFrame.")
207	self._fitted = 1	4✔
208
209	def _predict(self, df):	4✔
210	if self._need_fit and (self._fitted == 0):	4✔
211	raise RuntimeError("The model must be trained first.")
212	if isinstance(df, pd.DataFrame):	4✔
213	if df.columns.duplicated().any():	4✔
214	raise ValueError(
215	"Input DataFrame must have unique column names."
216	)
217	df_copy = df.copy()	4✔
218	predicted = self._predict_core(df_copy)	4✔
219	else:
220	raise TypeError("Input must be a pandas DataFrame.")
221	# make sure index freq is the same (because pandas has a bug that some
222	# operation, e.g. concat, may change freq)
223	predicted.index.freq = df.index.freq	4✔
224	return predicted	4✔
225
226	@abstractmethod	4✔
227	def _fit_core(self, df):	1✔
228	pass
229
230	@abstractmethod	4✔
231	def _predict_core(self, df):	1✔
232	pass
233
234	@abstractmethod	4✔
235	def fit(self, df):	1✔
236	pass
237
238	@abstractmethod	4✔
239	def predict(self, df):	1✔
240	pass
241
242	@abstractmethod	4✔
243	def fit_predict(self, df):	1✔
244	pass

arundo / adtk / 203

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