Source code for featuretools.entityset.timedelta

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")