survivalist.util

  1# This program is free software: you can redistribute it and/or modify
  2# it under the terms of the GNU General Public License as published by
  3# the Free Software Foundation, either version 3 of the License, or
  4# (at your option) any later version.
  5#
  6# This program is distributed in the hope that it will be useful,
  7# but WITHOUT ANY WARRANTY; without even the implied warranty of
  8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  9# GNU General Public License for more details.
 10#
 11# You should have received a copy of the GNU General Public License
 12# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 13import numpy as np
 14import pandas as pd
 15from pandas.api.types import CategoricalDtype
 16from sklearn.utils import check_array, check_consistent_length
 17
 18__all__ = ["check_array_survival", "check_y_survival", "safe_concat", "Surv"]
 19
 20
 21class Surv:
 22    """
 23    Helper class to construct structured array of event indicator and observed time.
 24    """
 25
 26    @staticmethod
 27    def from_arrays(event, time, name_event=None, name_time=None):
 28        """Create structured array.
 29
 30        Parameters
 31        ----------
 32        event : array-like
 33            Event indicator. A boolean array or array with values 0/1.
 34        time : array-like
 35            Observed time.
 36        name_event : str|None
 37            Name of event, optional, default: 'event'
 38        name_time : str|None
 39            Name of observed time, optional, default: 'time'
 40
 41        Returns
 42        -------
 43        y : np.array
 44            Structured array with two fields.
 45        """
 46        name_event = name_event or "event"
 47        name_time = name_time or "time"
 48        if name_time == name_event:
 49            raise ValueError("name_time must be different from name_event")
 50
 51        time = np.asanyarray(time, dtype=float)
 52        y = np.empty(time.shape[0], dtype=[
 53                     (name_event, bool), (name_time, float)])
 54        y[name_time] = time
 55
 56        event = np.asanyarray(event)
 57        check_consistent_length(time, event)
 58
 59        if np.issubdtype(event.dtype, np.bool_):
 60            y[name_event] = event
 61        else:
 62            events = np.unique(event)
 63            events.sort()
 64            if len(events) != 2:
 65                raise ValueError("event indicator must be binary")
 66
 67            if np.all(events == np.array([0, 1], dtype=events.dtype)):
 68                y[name_event] = event.astype(bool)
 69            else:
 70                raise ValueError(
 71                    "non-boolean event indicator must contain 0 and 1 only")
 72
 73        return y
 74
 75    @staticmethod
 76    def from_dataframe(event, time, data):
 77        """Create structured array from data frame.
 78
 79        Parameters
 80        ----------
 81        event : object
 82            Identifier of column containing event indicator.
 83        time : object
 84            Identifier of column containing time.
 85        data : pandas.DataFrame
 86            Dataset.
 87
 88        Returns
 89        -------
 90        y : np.array
 91            Structured array with two fields.
 92        """
 93        if not isinstance(data, pd.DataFrame):
 94            raise TypeError(
 95                f"expected pandas.DataFrame, but got {type(data)!r}")
 96
 97        return Surv.from_arrays(
 98            data.loc[:, event].values,
 99            data.loc[:, time].values,
100            name_event=str(event),
101            name_time=str(time),
102        )
103
104
105def check_y_survival(y_or_event, *args, allow_all_censored=False, allow_time_zero=True):
106    """Check that array correctly represents an outcome for survival analysis.
107
108    Parameters
109    ----------
110    y_or_event : structured array with two fields, or boolean array
111        If a structured array, it must contain the binary event indicator
112        as first field, and time of event or time of censoring as
113        second field. Otherwise, it is assumed that a boolean array
114        representing the event indicator is passed.
115
116    *args : list of array-likes
117        Any number of array-like objects representing time information.
118        Elements that are `None` are passed along in the return value.
119
120    allow_all_censored : bool, optional, default: False
121        Whether to allow all events to be censored.
122
123    allow_time_zero : bool, optional, default: True
124        Whether to allow event times to be zero.
125
126    Returns
127    -------
128    event : array, shape=[n_samples,], dtype=bool
129        Binary event indicator.
130
131    time : array, shape=[n_samples,], dtype=float
132        Time of event or censoring.
133    """
134    if len(args) == 0:
135        y = y_or_event
136
137        if not isinstance(y, np.ndarray) or y.dtype.fields is None or len(y.dtype.fields) != 2:
138            raise ValueError(
139                "y must be a structured array with the first field"
140                " being a binary class event indicator and the second field"
141                " the time of the event/censoring"
142            )
143
144        event_field, time_field = y.dtype.names
145        y_event = y[event_field]
146        time_args = (y[time_field],)
147    else:
148        y_event = np.asanyarray(y_or_event)
149        time_args = args
150
151    event = check_array(y_event, ensure_2d=False)
152    if not np.issubdtype(event.dtype, np.bool_):
153        raise ValueError(
154            f"elements of event indicator must be boolean, but found {event.dtype}")
155
156    if not (allow_all_censored or np.any(event)):
157        raise ValueError("all samples are censored")
158
159    return_val = [event]
160    for i, yt in enumerate(time_args):
161        if yt is None:
162            return_val.append(yt)
163            continue
164
165        yt = check_array(yt, ensure_2d=False)
166        if not np.issubdtype(yt.dtype, np.number):
167            raise ValueError(
168                f"time must be numeric, but found {yt.dtype} for argument {i + 2}")
169
170        if allow_time_zero:
171            cond = yt < 0
172            msg = "observed time contains values smaller zero"
173        else:
174            cond = yt <= 0
175            msg = "observed time contains values smaller or equal to zero"
176        if np.any(cond):
177            raise ValueError(msg)
178
179        return_val.append(yt)
180
181    return tuple(return_val)
182
183
184def check_array_survival(X, y, **kwargs):
185    """Check that all arrays have consistent first dimensions.
186
187    Parameters
188    ----------
189    X : array-like
190        Data matrix containing feature vectors.
191
192    y : structured array with two fields
193        A structured array containing the binary event indicator
194        as first field, and time of event or time of censoring as
195        second field.
196
197    kwargs : dict
198        Additional arguments passed to :func:`check_y_survival`.
199
200    Returns
201    -------
202    event : array, shape=[n_samples,], dtype=bool
203        Binary event indicator.
204
205    time : array, shape=[n_samples,], dtype=float
206        Time of event or censoring.
207    """
208    event, time = check_y_survival(y, **kwargs)
209    check_consistent_length(X, event, time)
210    return event, time
211
212
213def safe_concat(objs, *args, **kwargs):
214    """Alternative to :func:`pandas.concat` that preserves categorical variables.
215
216    Parameters
217    ----------
218    objs : a sequence or mapping of Series, DataFrame, or Panel objects
219        If a dict is passed, the sorted keys will be used as the `keys`
220        argument, unless it is passed, in which case the values will be
221        selected (see below). Any None objects will be dropped silently unless
222        they are all None in which case a ValueError will be raised
223    axis : {0, 1, ...}, default 0
224        The axis to concatenate along
225    join : {'inner', 'outer'}, default 'outer'
226        How to handle indexes on other axis(es)
227    join_axes : list of Index objects
228        Specific indexes to use for the other n - 1 axes instead of performing
229        inner/outer set logic
230    verify_integrity : boolean, default False
231        Check whether the new concatenated axis contains duplicates. This can
232        be very expensive relative to the actual data concatenation
233    keys : sequence, default None
234        If multiple levels passed, should contain tuples. Construct
235        hierarchical index using the passed keys as the outermost level
236    levels : list of sequences, default None
237        Specific levels (unique values) to use for constructing a
238        MultiIndex. Otherwise they will be inferred from the keys
239    names : list, default None
240        Names for the levels in the resulting hierarchical index
241    ignore_index : boolean, default False
242        If True, do not use the index values along the concatenation axis. The
243        resulting axis will be labeled 0, ..., n - 1. This is useful if you are
244        concatenating objects where the concatenation axis does not have
245        meaningful indexing information. Note the the index values on the other
246        axes are still respected in the join.
247    copy : boolean, default True
248        If False, do not copy data unnecessarily
249
250    Notes
251    -----
252    The keys, levels, and names arguments are all optional
253
254    Returns
255    -------
256    concatenated : type of objects
257    """
258    axis = kwargs.pop("axis", 0)
259    categories = {}
260    for df in objs:
261        if isinstance(df, pd.Series):
262            if isinstance(df.dtype, CategoricalDtype):
263                categories[df.name] = {
264                    "categories": df.cat.categories,
265                    "ordered": df.cat.ordered,
266                }
267        else:
268            dfc = df.select_dtypes(include=["category"])
269            for name, s in dfc.items():
270                if name in categories:
271                    if axis == 1:
272                        raise ValueError(f"duplicate columns {name}")
273                    if not categories[name]["categories"].equals(s.cat.categories):
274                        raise ValueError(
275                            f"categories for column {name} do not match")
276                else:
277                    categories[name] = {
278                        "categories": s.cat.categories,
279                        "ordered": s.cat.ordered,
280                    }
281                df[name] = df[name].astype(object)
282
283    concatenated = pd.concat(objs, *args, axis=axis, **kwargs)
284
285    for name, params in categories.items():
286        concatenated[name] = pd.Categorical(concatenated[name], **params)
287
288    return concatenated
289
290
291class _PropertyAvailableIfDescriptor:
292    """Implements a conditional property using the descriptor protocol based on the property decorator.
293
294    The corresponding class in scikit-learn (`_AvailableIfDescriptor`) only supports callables.
295    This class adopts the property decorator as described in the descriptor guide in the offical Python documentation.
296
297    See also
298    --------
299    https://docs.python.org/3/howto/descriptor.html
300        Descriptor HowTo Guide
301
302    :class:`sklearn.utils.available_if._AvailableIfDescriptor`
303        The original class in scikit-learn.
304    """
305
306    def __init__(self, check, fget, doc=None):
307        self.check = check
308        self.fget = fget
309        if doc is None and fget is not None:
310            doc = fget.__doc__
311        self.__doc__ = doc
312        self._name = ""
313
314    def __set_name__(self, owner, name):
315        self._name = name
316
317    def __get__(self, obj, objtype=None):
318        if obj is None:
319            return self
320
321        attr_err = AttributeError(
322            f"This {obj!r} has no attribute {self._name!r}")
323        if not self.check(obj):
324            raise attr_err
325
326        if self.fget is None:
327            raise AttributeError(f"property '{self._name}' has no getter")
328        return self.fget(obj)
329
330
331def property_available_if(check):
332    """A property attribute that is available only if check returns a truthy value.
333
334    Only supports getting an attribute value, setting or deleting an attribute value are not supported.
335
336    Parameters
337    ----------
338    check : callable
339        When passed the object of the decorated method, this should return
340        `True` if the property attribute is available, and either return `False`
341        or raise an `AttributeError` if not available.
342
343    Returns
344    -------
345    callable
346        Callable makes the decorated property available if `check` returns
347        `True`, otherwise the decorated property is unavailable.
348    """
349    return lambda fn: _PropertyAvailableIfDescriptor(check=check, fget=fn)