NOTICE
The upcoming release of Featuretools 1.0.0 contains several breaking changes. Users are encouraged to test this version prior to release by installing from GitHub:
pip install https://github.com/alteryx/featuretools/archive/woodwork-integration.zip
For details on migrating to the new version, refer to Transitioning to Featuretools Version 1.0. Please report any issues in the Featuretools GitHub repo or by messaging in Alteryx Open Source Slack.
import pandas as pd from dateutil.relativedelta import relativedelta [docs]class Timedelta(object): """Represents differences in time. Timedeltas can be defined in multiple units. Supported units: - "ms" : milliseconds - "s" : seconds - "h" : hours - "m" : minutes - "d" : days - "o"/"observations" : number of individual events - "mo" : months - "Y" : years Timedeltas can also be defined in terms of observations. In this case, the Timedelta represents the period spanned by `value`. For observation timedeltas: >>> three_observations_log = Timedelta(3, "observations") >>> three_observations_log.get_name() '3 Observations' """ _Observations = "o" # units for absolute times _absolute_units = ['ms', 's', 'h', 'm', 'd', 'w'] _relative_units = ['mo', 'Y'] _readable_units = { "ms": "Milliseconds", "s": "Seconds", "h": "Hours", "m": "Minutes", "d": "Days", "o": "Observations", "w": "Weeks", "Y": "Years", "mo": "Months" } _readable_to_unit = {v.lower(): k for k, v in _readable_units.items()} [docs] def __init__(self, value, unit=None, delta_obj=None): """ Args: value (float, str, dict) : Value of timedelta, string providing both unit and value, or a dictionary of units and times. unit (str) : Unit of time delta. delta_obj (pd.Timedelta or pd.DateOffset) : A time object used internally to do time operations. If None is provided, one will be created using the provided value and unit. """ self.check_value(value, unit) self.times = self.fix_units() if delta_obj is not None: self.delta_obj = delta_obj else: self.delta_obj = self.get_unit_type() @classmethod def from_dictionary(cls, dictionary): dict_units = dictionary['unit'] dict_values = dictionary['value'] if isinstance(dict_units, str) and isinstance(dict_values, (int, float)): return cls({dict_units: dict_values}) else: all_units = dict() for i in range(len(dict_units)): all_units[dict_units[i]] = dict_values[i] return cls(all_units) @classmethod def make_singular(cls, s): if len(s) > 1 and s.endswith('s'): return s[:-1] return s @classmethod def _check_unit_plural(cls, s): if len(s) > 2 and not s.endswith('s'): return (s + 's').lower() elif len(s) > 1: return s.lower() return s def get_value(self, unit=None): if unit is not None: return self.times[unit] elif len(self.times.values()) == 1: return list(self.times.values())[0] else: return self.times def get_units(self): return list(self.times.keys()) def get_unit_type(self): all_units = self.get_units() if self._Observations in all_units: return None elif self.is_absolute() and self.has_multiple_units() is False: return pd.Timedelta(self.times[all_units[0]], all_units[0]) else: readable_times = self.lower_readable_times() return relativedelta(**readable_times) def check_value(self, value, unit): if isinstance(value, str): from featuretools.utils.wrangle import _check_timedelta td = _check_timedelta(value) self.times = td.times elif isinstance(value, dict): self.times = value else: self.times = {unit: value} def fix_units(self): fixed_units = dict() for unit, value in self.times.items(): unit = self._check_unit_plural(unit) if unit in self._readable_to_unit: unit = self._readable_to_unit[unit] fixed_units[unit] = value return fixed_units def lower_readable_times(self): readable_times = dict() for unit, value in self.times.items(): readable_unit = self._readable_units[unit].lower() readable_times[readable_unit] = value return readable_times def get_name(self): all_units = self.get_units() if self.has_multiple_units() is False: return "{} {}".format(self.times[all_units[0]], self._readable_units[all_units[0]]) final_str = "" for unit, value in self.times.items(): if value == 1: unit = self.make_singular(unit) final_str += "{} {} ".format(value, self._readable_units[unit]) return final_str[:-1] def get_arguments(self): units = list() values = list() for unit, value in self.times.items(): units.append(unit) values.append(value) if len(units) == 1: return {'unit': units[0], 'value': values[0]} else: return {'unit': units, 'value': values} def is_absolute(self): for unit in self.get_units(): if unit not in self._absolute_units: return False return True def has_no_observations(self): for unit in self.get_units(): if unit in self._Observations: return False return True def has_multiple_units(self): if len(self.get_units()) > 1: return True else: return False def __eq__(self, other): if not isinstance(other, Timedelta): return False return (self.times == other.times) def __neg__(self): """Negate the timedelta""" new_times = dict() for unit, value in self.times.items(): new_times[unit] = -value if self.delta_obj is not None: return Timedelta(new_times, delta_obj=-self.delta_obj) else: return Timedelta(new_times) def __radd__(self, time): """Add the Timedelta to a timestamp value""" if self._Observations not in self.get_units(): return time + self.delta_obj else: raise Exception("Invalid unit") def __rsub__(self, time): """Subtract the Timedelta from a timestamp value""" if self._Observations not in self.get_units(): return time - self.delta_obj else: raise Exception("Invalid unit")