bandicoot-0.4.0/0000755000076500000240000000000012545051250014054 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/0000755000076500000240000000000012545051250016016 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/__init__.py0000755000076500000240000000040012545050666020136 0ustar cynddlstaff00000000000000__all__ = ['core', 'individual', 'spatial', 'network', 'helper', 'io', 'utils', 'tests', 'special'] from .io import read_csv from .core import User from . import individual, spatial, network, helper, utils, io, tests, core, special __version__ = "0.4.0" bandicoot-0.4.0/bandicoot/core.py0000755000076500000240000002677212545050666017353 0ustar cynddlstaff00000000000000from __future__ import division import datetime from collections import Counter from bandicoot.helper.tools import Colors from bandicoot.helper.group import _binning import bandicoot as bc class Record(object): """ Data structure storing a call detail record. Attributes ---------- interaction : str, 'text' or 'call' The type of interaction (text message or call). direction : str, 'in' or 'out' The direction of the record (incoming or outgoing). correspondent_id : str A unique identifier for the corresponding contact datetime : datetime The exact date and time of the interaction call_duration : int or None Durations of the call in seconds. None if the record is a text message. position : Position The geographic position of the user at the time of the interaction. """ __slots__ = ['interaction', 'direction', 'correspondent_id', 'datetime', 'call_duration', 'position'] def __init__(self, interaction, direction, correspondent_id, datetime, call_duration, position): self.interaction = interaction self.direction = direction self.correspondent_id = correspondent_id self.datetime = datetime self.call_duration = call_duration self.position = position def __repr__(self): return "Record(" + ", ".join(map(lambda x: "%s=%r" % (x, getattr(self, x)), self.__slots__)) + ")" def __eq__(self, other): if isinstance(other, self.__class__) and self.__slots__ == other.__slots__: return all(getattr(self, attr) == getattr(other, attr) for attr in self.__slots__) return False def __hash__(self): return hash(self.__repr__()) def matches(self, other): """ Returns true if two records 'match' - that is, they correspond to the same event from two perspectives. """ return self.interaction == other.interaction and \ self.direction != other.direction and \ self.call_duration == other.call_duration and \ abs((self.datetime - other.datetime).total_seconds()) < 30 def all_matches(self, iterable): return filter(self.matches, iterable) def has_match(self, iterable): return len(self.all_matches(iterable)) > 0 class Position(object): """ Data structure storing a generic location. Can be instantiated with either an antenna or a gps location. Printing out the position will show which was used to instantiate it. """ def __init__(self, antenna=None, location=None): self.antenna = antenna self.location = location def _get_location(self, user): if self.location: return self.location elif self.antenna: return user.antennas.get(self.antenna) else: return None def type(self): if self.antenna: return 'antenna' else: return 'gps' def __repr__(self): if self.antenna and self.location: return "Position(antenna=%s, location=%s)" % (self.antenna, self.location) if self.antenna: return "Position(antenna=%s)" % self.antenna if self.location: return "Position(location=(%s, %s))" % self.location return "Position()" def __eq__(self, other): if not isinstance(other, Position): return False if self.antenna and other.antenna: return self.antenna == other.antenna if self.location and other.location: return self.location == other.location if not self.location and not self.antenna and not other.location and not other.antenna: return True return False def __ne__(self, other): return not self.__eq__(other) def __hash__(self): return hash(self.__repr__()) class User(object): """ Data structure storing all the call, text or mobility records of the user. """ def __init__(self): self._records = [] self._antennas = {} self.name = None self.antennas_path = None self.attributes_path = None self.start_time = None self.end_time = None self.night_start = datetime.time(19) self.night_end = datetime.time(7) self.weekend = [6, 7] # Saturday, Sunday by default self.home = None self.has_text = False self.has_call = False self.has_antennas = False self.attributes = {} self.ignored_records = None self.percent_outofnetwork_calls = 0 self.percent_outofnetwork_texts = 0 self.percent_outofnetwork_contacts = 0 self.percent_outofnetwork_call_durations = 0 self.network = {} @property def antennas(self): """ The purpose of this is to hook into assignments to the user's antenna dictionary, and update records' location based on the new value. """ return self._antennas @antennas.setter def antennas(self, input_): self._antennas = input_ self.has_antennas = len(input_) > 0 for r in self._records: if r.position.antenna in self._antennas: r.position.location = self._antennas[r.position.antenna] else: r.position.location = None @property def records(self): """ Can be used to get or set the list of user's records. If the records are modified, the start and end time as well as the infered house properties of the object User are recomputed. """ return self._records @records.setter def records(self, input): self._records = sorted(input, key=lambda r: r.datetime) if len(self._records) > 0: self.start_time = self._records[0].datetime self.end_time = self._records[-1].datetime # Reset all the states self.has_call = False self.has_text = False self.has_antennas = False for r in self._records: if r.interaction == 'text': self.has_text = True elif r.interaction == 'call': self.has_call = True if r.position.type() == 'antenna': self.has_antennas = True self.recompute_home() def recompute_missing_neighbors(self): """ Recomputes statistics for missing users of the current user's network: - ``User.percent_outofnetwork_calls`` - ``User.percent_outofnetwork_texts`` - ``User.percent_outofnetwork_contacts`` - ``User.percent_outofnetwork_call_durations`` This function is automatically called from :meth:`~bandicoot.io.read_csv` when loading a network user. """ oon_records = [r for r in self.records if self.network.get(r.correspondent_id, None) is None] num_oon_calls = len([r for r in oon_records if r.interaction == 'call']) num_oon_texts = len([r for r in oon_records if r.interaction == 'text']) num_oon_neighbors = len(set(x.correspondent_id for x in oon_records)) oon_call_durations = sum([r.call_duration for r in oon_records if r.interaction == 'call']) num_calls = len([r for r in self.records if r.interaction == 'call']) num_texts = len([r for r in self.records if r.interaction == 'text']) total_neighbors = len(set(x.correspondent_id for x in self.records)) total_call_durations = sum([r.call_duration for r in self.records if r.interaction == 'call']) def _safe_div(a, b, default): return a / b if b != 0 else default # We set the percentage at 0 if no event occurs self.percent_outofnetwork_calls = _safe_div(num_oon_calls, num_calls, 0) self.percent_outofnetwork_texts = _safe_div(num_oon_texts, num_texts, 0) self.percent_outofnetwork_contacts = _safe_div(num_oon_neighbors, total_neighbors, 0) self.percent_outofnetwork_call_durations = _safe_div(oon_call_durations, total_call_durations, 0) def describe(self): """ Generate a short description of the object. Examples -------- >>> import bandicoot as bc >>> user = bc.User() >>> user.records = bc.tests.generate_user.random_burst(5) >>> user.describe() [x] 5 records from 2014-01-01 10:41:00 to 2014-01-01 11:21:00 5 contacts [x] 1 attribute Note ---- The summary is directly sent to the standard output. """ def format_int(name, n): if n == 0 or n == 1: return "%i %s" % (n, name[:-1]) else: return "%i %s" % (n, name) empty_box = Colors.OKGREEN + '[ ]' + Colors.ENDC + ' ' filled_box = Colors.OKGREEN + '[x]' + Colors.ENDC + ' ' if self.start_time is None: print empty_box + "No records stored" else: print (filled_box + format_int("records", len(self.records)) + " from %s to %s" % (self.start_time, self.end_time)) nb_contacts = bc.individual.number_of_contacts(self, interaction='callandtext', groupby=None) nb_contacts = nb_contacts['allweek']['allday']['callandtext'] if nb_contacts: print filled_box + format_int("contacts", nb_contacts) else: print empty_box + "No contacts" if self.has_attributes: print filled_box + format_int("attributes", len(self.attributes)) else: print empty_box + "No attribute stored" if len(self.antennas) == 0: print empty_box + "No antenna stored" else: print filled_box + format_int("antennas", len(self.antennas)) if self.has_home: print filled_box + "Has home" else: print empty_box + "No home" if self.has_text: print filled_box + "Has texts" else: print empty_box + "No texts" if self.has_call: print filled_box + "Has calls" else: print empty_box + "No calls" if self.has_network: print filled_box + "Has network" else: print empty_box + "No network" def recompute_home(self): """ Return the antenna where the user spends most of his time at night. None is returned if there are no candidates for a home antenna """ if self.night_start < self.night_end: night_filter = lambda r: self.night_end > r.datetime.time() > self.night_start else: night_filter = lambda r: not(self.night_end < r.datetime.time() < self.night_start) # Bin positions by chunks of 30 minutes candidates = list(_binning(filter(night_filter, self._records))) if len(candidates) == 0: self.home = None else: self.home = Counter(candidates).most_common()[0][0] return self.home @property def has_home(self): return self.home is not None @property def has_attributes(self): return len(self.attributes) != 0 @property def has_network(self): return self.network != {} def set_home(self, new_home): """ Sets the user's home. The argument can be a Position object or a tuple containing location data. """ if type(new_home) is Position: self.home = new_home elif type(new_home) is tuple: self.home = Position(location=new_home) else: self.home = Position(antenna=new_home) bandicoot-0.4.0/bandicoot/helper/0000755000076500000240000000000012545051250017275 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/helper/__init__.py0000644000076500000240000000000012545050666021406 0ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/helper/group.py0000644000076500000240000002625712545050666021031 0ustar cynddlstaff00000000000000from functools import partial import itertools from bandicoot.helper.tools import mean, std, SummaryStats, advanced_wrap, AutoVivification DATE_GROUPERS = { None: lambda _: None, "day": lambda d: d.isocalendar(), "week": lambda d: d.isocalendar()[0:2], "month": lambda d: (d.year, d.month), "year": lambda d: d.year } def _group_date(records, _fun): for _, chunk in itertools.groupby(records, key=lambda r: _fun(r.datetime)): yield chunk def group_records(user, interaction=None, groupby='week', part_of_week='allweek', part_of_day='allday'): """ Group records by year and week number. This function is used by the ``@grouping`` decorator. Parameters ---------- records : iterator An iterator over records groupby : Default is 'week': * 'week': group all records by year and week; * None: records are not grouped. This is useful if you don't want to divide records in chunks. * "day", "month", and "year" also accepted. part_of_week : {'allweek', 'weekday', 'weekend'}, default 'allweek' * 'weekend': keep only the weekend records * 'weekday': keep only the weekdays records * 'allweek': use all the records part_of_day : {'allday', 'day', 'night'}, default 'allday' * 'day': keep only the records during the day * 'night': keep only the records during the night * 'allday': use all the records interaction : object The interaction to filter records: * "callandtext", for only callandtext; * a string, to filter for one type; * None, to use all records. """ records = user.records if interaction == 'callandtext': records = filter(lambda r: r.interaction in ['call', 'text'], records) elif interaction is not None: records = filter(lambda r: r.interaction == interaction, records) if part_of_week == 'weekday': records = filter(lambda r: r.datetime.isoweekday() not in user.weekend, records) elif part_of_week == 'weekend': records = filter(lambda r: r.datetime.isoweekday() in user.weekend, records) elif part_of_week is not 'allweek': raise KeyError("{} is not a valid value for part_of_week. it should be 'weekday', 'weekend' or 'allweek'.".format(part_of_week)) if user.night_start < user.night_end: night_filter = lambda r: user.night_end > r.datetime.time() > user.night_start else: night_filter = lambda r: not(user.night_end < r.datetime.time() < user.night_start) if part_of_day == 'day': records = filter(lambda r: not(night_filter(r)), records) elif part_of_day == 'night': records = filter(night_filter, records) elif part_of_day is not 'allday': raise KeyError("{} is not a valid value for part_of_day. It should be 'day', 'night' or 'allday'.".format(part_of_day)) return _group_date(records, DATE_GROUPERS[groupby]) def statistics(data, summary='default', datatype=None): """ Return statistics (mean, standard error, standard error and median, min and max) on data metrics. Examples -------- Given a list of integers or floating point numbers, ``statistics`` computes the mean and standard error of the mean, and the min and max. >>> statistics([0, 1, 2, 3]) {'mean': 1.5, 'std': 1.2910, 'min': 0, 'max': 3} Given a list of ``SummaryStats`` tuples, the function will returns the mean, standard error of the mean, min and max for each attribute of the tuples. """ def _default_stats(agg): if agg is None or len(agg) == 0: return {'mean': None, 'std': None} else: # Some functions may return None values # It's better to filter them agg = filter(lambda x: x is not None, agg) return {'mean': mean(agg), 'std': std(agg)} def _stats_dict(v): return {key: _default_stats([getattr(s, key, None) for s in data]) for key in v} summary_keys = { 'default': ['mean', 'std'], 'extended': ['mean', 'std', 'median', 'skewness', 'kurtosis', 'min', 'max'] } # Infer the data type of 'data' if datatype is None: if isinstance(data, (type(None), int, float)): datatype = 'scalar' elif isinstance(data, SummaryStats): datatype = 'summarystats' elif hasattr(data, "__len__"): # list or numpy array data = filter(lambda x: x is not None, data) if len(data) == 0 or isinstance(data[0], (int, float)): datatype = 'distribution_scalar' elif isinstance(data[0], SummaryStats): datatype = 'distribution_summarystats' else: raise TypeError("{} is not a valid input. It should be a number, a SummaryStats object, or None".format(data[0])) else: raise TypeError("{} is not a valid input. It should be a number, a SummaryStats object, or a list".format(data)) if datatype == 'scalar': return data if datatype == 'summarystats': if summary is None: return data.distribution elif summary in ['default', 'extended']: return {key: getattr(data, key, None) for key in summary_keys[summary]} else: raise ValueError("{} is not a valid summary type".format(summary)) if datatype == 'distribution_scalar': if summary is 'default': return _default_stats(data) elif summary is None: return data else: raise ValueError("{} is not a valid summary type".format(summary)) if datatype == 'distribution_summarystats': if summary is None: return [item.distribution for item in data] elif summary in ['extended', 'default']: return _stats_dict(summary_keys[summary]) else: raise ValueError("{} is not a valid summary type".format(summary)) else: raise ValueError("{} is not a valid data type.".format(datatype)) def grouping(f=None, user_kwd=False, interaction=['call', 'text'], summary='default'): """ ``grouping`` is a decorator for indicator functions, used to simplify the source code. Parameters ---------- f : function The function to decorate user_kwd : boolean If user_kwd is True, the user object will be passed to the decorated function interaction : 'call', 'text', 'location', or a list By default, all indicators use only 'call' and 'text' records, but the interaction keywords filters the records passed to the function. summary: 'default', 'extended', None An indicator returns data statistics, ether *mean* and *std* by default, more with 'extended', or the inner distribution with None. See :meth:`~bandicoot.helper.group.statistics` for more details. See :ref:`new-indicator-label` to learn how to write an indicator with this decorator. """ if f is None: return partial(grouping, user_kwd=user_kwd, interaction=interaction, summary=summary) def wrapper(user, groupby='week', interaction=interaction, summary=summary, split_week=False, split_day=False, datatype=None, **kwargs): if interaction is None: interaction = ['call', 'text'] elif isinstance(interaction, str): interaction = [interaction] else: interaction = interaction[:] # copy the list for more safety part_of_day = ['allday'] if split_day: part_of_day += ['day', 'night'] part_of_week = ['allweek'] if split_week: part_of_week += ['weekday', 'weekend'] for i in interaction: if i not in ['callandtext', 'call', 'text', 'location']: raise ValueError("%s is not a valid interaction value. Only 'call', " "'text', and 'location' are accepted." % i) def map_filters(interaction, part_of_week, part_of_day): """ Call the wrapped function for every combinations of interaction, part_of_week, and part_of_day. """ for i in interaction: for filter_week in part_of_week: for filter_day in part_of_day: if user_kwd is True: result = [f(g, user, **kwargs) for g in group_records(user, i, groupby, filter_week, filter_day)] else: result = [f(g, **kwargs) for g in group_records(user, i, groupby, filter_week, filter_day)] yield filter_week, filter_day, i, result returned = AutoVivification() # nested dict structure for (f_w, f_d, i, m) in map_filters(interaction, part_of_week, part_of_day): if groupby is None: m = m[0] if len(m) != 0 else None returned[f_w][f_d][i] = statistics(m, summary=summary, datatype=datatype) return returned return advanced_wrap(f, wrapper) def _binning(records): """ Bin records by chunks of 30 minutes, returning the most prevalent position. """ records = list(records) from collections import Counter def get_key(d): from datetime import datetime, timedelta k = d + timedelta(minutes=-(d.minute % 30)) return datetime(k.year, k.month, k.day, k.hour, k.minute, 0) chunks = itertools.groupby(records, key=lambda r: get_key(r.datetime)) for _, items in chunks: positions = [i.position for i in items] yield Counter(positions).most_common(1)[0][0] def spatial_grouping(f=None, user_kwd=False, summary='default', use_records=False): if f is None: return partial(spatial_grouping, user_kwd=user_kwd, summary=summary, use_records=use_records) if use_records is True: map_records = lambda x: x else: map_records = _binning def wrapper(user, groupby='week', summary=summary, split_week=False, split_day=False, datatype=None, **kwargs): part_of_day = ['allday'] if split_day: part_of_day += ['day', 'night'] part_of_week = ['allweek'] if split_week: part_of_week += ['weekday', 'weekend'] def map_filters_spatial(part_of_week, part_of_day): """ Call the wrapped function for every combinations of interaction, part_of_week, and part_of_day. """ for filter_week in part_of_week: for filter_day in part_of_day: if user_kwd is True: result = [f(map_records(g), user, **kwargs) for g in group_records(user, None, groupby, filter_week, filter_day)] else: result = [f(map_records(g), **kwargs) for g in group_records(user, None, groupby, filter_week, filter_day)] yield filter_week, filter_day, result returned = AutoVivification() # nested dict structure for (f_w, f_d, m) in map_filters_spatial(part_of_week, part_of_day): if groupby is None: m = m[0] if len(m) != 0 else None returned[f_w][f_d] = statistics(m, summary=summary, datatype=datatype) return returned return advanced_wrap(f, wrapper) bandicoot-0.4.0/bandicoot/helper/tools.py0000755000076500000240000002062612545050672021027 0ustar cynddlstaff00000000000000from collections import OrderedDict as NativeOrderedDict from functools import update_wrapper import itertools import inspect import math import json try: from thread import get_ident as _get_ident except ImportError: from dummy_thread import get_ident as _get_ident class CustomEncoder(json.JSONEncoder): def default(self, obj): from bandicoot.core import User if isinstance(obj, User): return repr(obj) return json.JSONEncoder.default(self, obj) class OrderedDict(NativeOrderedDict): def __repr__(self): if not self: return '%s()' % (self.__class__.__name__,) # This solution is easy to implement but not robust s = json.dumps(self, cls=CustomEncoder, indent=4) transformations = [ ('"<', '<'), ('>"', '>'), ('null', 'None') ] for old_pattern, new_pattern in transformations: s = s.replace(old_pattern, new_pattern) return s def advanced_wrap(f, wrapper): """ Wrap a decorated function while keeping the same keyword arguments """ f_sig = list(inspect.getargspec(f)) wrap_sig = list(inspect.getargspec(wrapper)) # Update the keyword arguments of the wrapper if f_sig[3] is None or f_sig[3] == []: f_sig[3], f_kwargs = [], [] else: f_kwargs = f_sig[0][-len(f_sig[3]):] for key, default in zip(f_kwargs, f_sig[3]): wrap_sig[0].append(key) wrap_sig[3] = wrap_sig[3] + (default, ) wrap_sig[2] = None # Remove kwargs src = "lambda %s: " % (inspect.formatargspec(*wrap_sig)[1:-1]) new_args = inspect.formatargspec( wrap_sig[0], wrap_sig[1], wrap_sig[2], f_kwargs, formatvalue=lambda x: '=' + x) src += 'wrapper%s\n' % new_args decorated = eval(src, locals()) return update_wrapper(decorated, f) class Colors: """ The `Colors` class stores six codes to color a string. It can be used to print messages inside a terminal prompt. Examples -------- >>> print Colors.FAIL + "Error: it's a failure!" + Colors.ENDC Attributes ---------- HEADER Header color. OKBLUE Blue color for a success message OKGREEN Green color for a success message WARNING Warning color (yellow) FAIL Failing color (red) ENDC Code to disable coloring. Always add it after coloring a string. """ def __init__(self): pass HEADER = '\033[95m' OKBLUE = '\033[94m' OKGREEN = '\033[32m' WARNING = '\033[33m' FAIL = '\033[31m' ENDC = '\033[0m' def warning_str(str): """ Return a colored string using the warning color (`Colors.WARNING`). """ return Colors.WARNING + str + Colors.ENDC def percent_records_missing_location(user, method=None): """ Return the percentage of records missing a location parameter. """ if len(user.records) == 0: return 0. missing_locations = sum([1 for record in user.records if record.position._get_location(user) is None]) return float(missing_locations) / len(user.records) def antennas_missing_locations(user, Method=None): unique_antennas = set([record.position.antenna for record in user.records if record.position.antenna is not None]) return sum([1 for antenna in unique_antennas if user.antennas.get(antenna) is None]) def pairwise(iterable): "s -> (s0,s1), (s1,s2), (s2, s3), ..." a, b = itertools.tee(iterable) next(b, None) return itertools.izip(a, b) def mean(data): """ Return the arithmetic mean of ``data``. Examples -------- >>> mean([1, 2, 3, 4, 4]) 2.8 """ if len(data) < 1: return None return float(sum(data)) / len(data) def kurtosis(data): """ Return the kurtosis for ``data``. """ if len(data) == 0: return None num = moment(data, 4) denom = moment(data, 2) ** 2. return num / denom if denom != 0 else 0 def skewness(data): """ Returns the skewness of ``data``. """ if len(data) == 0: return None num = moment(data, 3) denom = moment(data, 2) ** 1.5 return num / denom if denom != 0 else 0. def std(data): if len(data) == 0: return None variance = moment(data, 2) return variance ** 0.5 def moment(data, n): if len(data) <= 1: return 0 _mean = mean(data) return float(sum([(item - _mean) ** n for item in data])) / len(data) def median(data): """ Return the median of numeric data, unsing the "mean of middle two" method. If ``data`` is empty, ``0`` is returned. Examples -------- >>> median([1, 3, 5]) 3.0 When the number of data points is even, the median is interpolated: >>> median([1, 3, 5, 7]) 4.0 """ if len(data) == 0: return None data.sort() return float((data[len(data) / 2] + data[(len(data) - 1) / 2]) / 2.) def minimum(data): if len(data) == 0: return None return float(min(data)) def maximum(data): if len(data) == 0: return None return float(max(data)) class SummaryStats(object): """ Data structure storing a numeric distribution Attributes ---------- mean : float Mean of the distribution. std : float The standard deviation of the distribution. min: float The minimum value of the distribution. max: float The max value of the distribution. median : float The median value of the distribution skewness : float The skewness of the distribution, measuring its asymmetry kurtosis : float The kurtosis of the distribution, measuring its "peakedness" distribution : list The complete distribution, as a list of floats Note ---- You can generate a *SummaryStats* object using the :meth:`~bandicoot.helper.tools.summary_stats` function. """ __slots__ = ['mean', 'std', 'min', 'max', 'median', 'skewness', 'kurtosis', 'distribution'] def __init__(self, mean, std, min, max, median, skewness, kurtosis, distribution): self.mean, self.std, self.min, self.max, self.median, self.skewness, self.kurtosis, self.distribution = mean, std, min, max, median, skewness, kurtosis, distribution def __repr__(self): return "SummaryStats(" + ", ".join(map(lambda x: "%s=%r" % (x, getattr(self, x)), self.__slots__)) + ")" def __eq__(self, other): if isinstance(other, self.__class__) and self.__slots__ == other.__slots__: return all(getattr(self, attr) == getattr(other, attr) for attr in self.__slots__) return False def summary_stats(data): """ Returns a :class:`~bandicoot.helper.tools.SummaryStats` object containing informations on the given distribution. Example ------- >>> summary_stats([0, 1]) SummaryStats(mean=0.5, std=0.5, min=0.0, max=1.0, median=0.5, skewness=0.0, kurtosis=1.0, distribution=[0, 1]) """ if len(data) < 1: return SummaryStats(0., 0., 0., 0., 0., 0., 0., []) data.sort() _median = median(data) _mean = mean(data) _std = std(data) _minimum = minimum(data) _maximum = maximum(data) _kurtosis = kurtosis(data) _skewness = skewness(data) _distribution = data return SummaryStats(_mean, _std, _minimum, _maximum, _median, _skewness, _kurtosis, _distribution) def entropy(data): """ Compute the Shannon entropy, a measure of uncertainty. """ if len(data) == 0: return None n = sum(data) _op = lambda f: f * math.log(f) return - sum(_op(float(i) / n) for i in data) def great_circle_distance(pt1, pt2): r = 6371. delta_latitude = (pt1[0] - pt2[0]) / 180 * math.pi delta_longitude = (pt1[1] - pt2[1]) / 180 * math.pi latitude1 = pt1[0] / 180 * math.pi latitude2 = pt2[0] / 180 * math.pi return r * 2. * math.asin(math.sqrt(math.pow(math.sin(delta_latitude / 2), 2) + math.cos(latitude1) * math.cos(latitude2) * math.pow(math.sin(delta_longitude / 2), 2))) class AutoVivification(dict): """ Implementation of perl's autovivification feature. Under CC-BY-SA 3.0 from nosklo on stackoverflow: http://stackoverflow.com/questions/19189274/defaultdict-of-defaultdict-nested """ def __getitem__(self, item): try: return dict.__getitem__(self, item) except KeyError: value = self[item] = type(self)() return value bandicoot-0.4.0/bandicoot/individual.py0000755000076500000240000003035512545050672020540 0ustar cynddlstaff00000000000000from __future__ import division from bandicoot.helper.group import grouping from bandicoot.helper.tools import summary_stats, entropy, pairwise from collections import Counter import math import datetime from collections import defaultdict @grouping def interevent_time(records): """ The interevent time between two records of the user. """ inter_events = pairwise(r.datetime for r in records) inter = [(new - old).total_seconds() for old, new in inter_events] return summary_stats(inter) @grouping def number_of_contacts(records, direction=None, more=0): """ The number of contacts the user interacted with. Parameters ---------- direction : str, optional Filters the records by their direction: ``None`` for all records, ``'in'`` for incoming, and ``'out'`` for outgoing. more : int, optional Counts only contacts with more than this number of interactions. Defaults to 0. """ if direction is None: counter = Counter(r.correspondent_id for r in records) else: counter = Counter(r.correspondent_id for r in records if r.direction == direction) return sum(1 for d in counter.values() if d > more) @grouping def entropy_of_contacts(records, normalize=False): """ The entropy of the user's contacts. Parameters ---------- normalize: boolean, default is False Returns a normalized entropy between 0 and 1. """ counter = Counter(r.correspondent_id for r in records) raw_entropy = entropy(counter.values()) n = len(counter) if normalize and n > 1: return raw_entropy / math.log(n) else: return raw_entropy @grouping def interactions_per_contact(records, direction=None): """ The number of interactions a user had with each of its contacts. Parameters ---------- direction : str, optional Filters the records by their direction: ``None`` for all records, ``'in'`` for incoming, and ``'out'`` for outgoing. """ if direction is None: counter = Counter(r.correspondent_id for r in records) else: counter = Counter(r.correspondent_id for r in records if r.direction == direction) return summary_stats(counter.values()) @grouping(user_kwd=True, interaction='call') def percent_initiated_interactions(records, user): """ The percentage of calls initiated by the user. """ records = list(records) if len(records) == 0: return 0 initiated = sum(1 for r in records if r.direction == 'out') return float(initiated) / len(records) @grouping(user_kwd=True) def percent_nocturnal(records, user): """ The percentage of interactions the user had at night. By default, nights are 7pm-7am. Nightimes can be set in ``User.night_start`` and ``User.night_end``. """ records = list(records) if len(records) == 0: return 0 if user.night_start < user.night_end: night_filter = lambda d: user.night_end > d.time() > user.night_start else: night_filter = lambda d: not(user.night_end < d.time() < user.night_start) return float(sum(1 for r in records if night_filter(r.datetime))) / len(records) @grouping(interaction='call') def call_duration(records, direction=None): """ The duration of the user's calls. Parameters ---------- direction : str, optional Filters the records by their direction: ``None`` for all records, ``'in'`` for incoming, and ``'out'`` for outgoing. """ if direction is None: call_durations = [r.call_duration for r in records] else: call_durations = [r.call_duration for r in records if r.direction == direction] return summary_stats(call_durations) def _conversations(group, delta=datetime.timedelta(hours=1)): """ Group texts into conversations. The function returns an iterator over records grouped by conversations. See :ref:`Using bandicoot ` for a definition of conversations. A conversation begins when one person sends a text-message to the other and ends when one of them makes a phone call or there is no activity between them for an hour. """ last_time = None results = [] for g in group: if last_time is None or g.datetime - last_time < delta: if g.interaction == 'text': results.append(g) # A call always ends a conversation else: if len(results) != 0: yield results results = [] else: if len(results) != 0: yield results if g.interaction == 'call': results = [] else: results = [g] last_time = g.datetime if len(results) != 0: yield results @grouping(interaction='callandtext') def response_rate_text(records): """ The response rate of the user (between 0 and 1). Considers text-conversations which began with an incoming text. Response rate is the fraction of such conversations in which the user sent a text (a response). The following sequence of messages defines four conversations (``I`` for an incoming text, ``O`` for an outgoing text): :: I-O-I-O => Started with an incoming text and at least one outgoing text I-I-O-I => Started with an incoming text and at least one outgoing text I-I-I-I => Started with an incoming text but doesn't have outgoing texts O-O-I-O => Not starting with an incoming text Here, the ratio would be 2/3 as we have 3 conversations starting with an incoming text and 2 of them have at least one outgoing text. See :ref:`Using bandicoot ` for a definition of conversations. """ records = list(records) interactions = defaultdict(list) for r in records: interactions[r.correspondent_id].append(r) def _response_rate(grouped): received, responded = 0, 0 conversations = _conversations(grouped) for conv in conversations: if len(conv) != 0: first = conv[0] if first.direction == 'in' and first.interaction == 'text': received += 1 if any((i.direction == 'out' for i in conv)): responded += 1 return responded, received # Group all records by their correspondent, and compute the response rate # for each all_couples = map(_response_rate, interactions.values()) responded, received = map(sum, zip(*all_couples)) return float(responded) / received if received != 0 else 0 @grouping(interaction='callandtext') def response_delay_text(records): """ The response delay of the user within a conversation (in seconds) The following sequence of messages defines conversations (``I`` for an incoming text, ``O`` for an outgoing text, ``-`` for a one minute delay): :: I-O--I----O, we have a 60 seconds response delay and a 240 seconds response delay O--O---I--O, we have a 1200 seconds response delay I--II---I-I, we don't have a response delay. The user hasn't answered For this user, the distribution of response delays will be ``[60, 240, 60]`` Notes ----- See :ref:`Using bandicoot ` for a definition of conversations. Conversation are defined to be a series of text messages each sent no more than an hour after the previous. The response delay can thus not be greater than one hour. """ records = list(records) interactions = defaultdict(list) for r in records: interactions[r.correspondent_id].append(r) def _response_delay(grouped): ts = ((b.datetime - a.datetime).total_seconds() for conv in _conversations(grouped) for a, b in pairwise(conv) if b.direction == 'out' and a.direction == 'in') return ts delays = [r for i in interactions.values() for r in _response_delay(i) if r > 0] if delays == []: return None return summary_stats(delays) @grouping(interaction='callandtext') def percent_initiated_conversations(records): """ The percentage of conversations that have been initiated by the user. Each call and each text conversation is weighted as a single interaction. See :ref:`Using bandicoot ` for a definition of conversations. """ records = list(records) interactions = defaultdict(list) for r in records: interactions[r.correspondent_id].append(r) def _percent_initiated(grouped): mapped = [(1 if conv[0].direction == 'out' else 0, 1) for conv in _conversations(grouped)] return mapped all_couples = [sublist for i in interactions.values() for sublist in _percent_initiated(i)] if len(all_couples) == 0: init, total = 0, 0 else: init, total = map(sum, zip(*all_couples)) return float(init) / total if total != 0 else 0 @grouping(interaction='callandtext') def active_days(records): """ The number of days during which the user was active. A user is considered active if he sends a text, receives a text, initiates a call, receives a call, or has a mobility point. """ days = set(r.datetime.date() for r in records) return len(days) @grouping def percent_pareto_interactions(records, percentage=0.8): """ The percentage of user's contacts that account for 80% of its interactions. """ records = list(records) if records == []: return 0. user_count = Counter(r.correspondent_id for r in records) target = int(math.ceil(sum(user_count.values()) * percentage)) user_sort = sorted(user_count.keys(), key=lambda x: user_count[x]) while target > 0 and len(user_sort) > 0: user_id = user_sort.pop() target -= user_count[user_id] return (len(user_count) - len(user_sort)) / len(records) @grouping(interaction='call') def percent_pareto_durations(records, percentage=0.8): """ The percentage of user's contacts that account for 80% of its total time spend on the phone. Optionally takes a percentage argument as a decimal (e.g., .8 for 80%). """ records = list(records) if records == []: return 0. user_count = defaultdict(int) for r in records: if r.interaction == "call": user_count[r.correspondent_id] += r.call_duration target = int(math.ceil(sum(user_count.values()) * percentage)) user_sort = sorted(user_count.keys(), key=lambda x: user_count[x]) while target > 0 and len(user_sort) > 0: user_id = user_sort.pop() target -= user_count[user_id] return (len(user_count) - len(user_sort)) / len(records) @grouping def balance_of_contacts(records, weighted=True): """ The balance of interactions per contact. For every contact, the balance is the number of outgoing interactions divided by the total number of interactions (in+out). .. math:: \\forall \\,\\text{contact}\\,c,\\;\\text{balance}\,(c) = \\frac{\\bigl|\\text{outgoing}\,(c)\\bigr|}{\\bigl|\\text{outgoing}\,(c)\\bigr|+\\bigl|\\text{incoming}\,(c)\\bigr|} Parameters ---------- weighted : str, optional If ``True``, the balance for each contact is weighted by the number of interactions the user had with this contact. """ counter_out = defaultdict(int) counter = defaultdict(int) for r in records: if r.direction == 'out': counter_out[r.correspondent_id] += 1 counter[r.correspondent_id] += 1 if not weighted: balance = [float(counter_out[c]) / float(counter[c]) for c in counter] else: balance = [float(counter_out[c]) / float(sum(counter.values())) for c in counter] return summary_stats(balance) @grouping() def number_of_interactions(records, direction=None): """ The number of interactions. Parameters ---------- direction : str, optional Filters the records by their direction: ``None`` for all records, ``'in'`` for incoming, and ``'out'`` for outgoing. """ if direction is None: return len([r for r in records]) else: return len([r for r in records if r.direction == direction]) bandicoot-0.4.0/bandicoot/io.py0000644000076500000240000005070412545050672017014 0ustar cynddlstaff00000000000000""" Contains tools for processing files (reading and writing csv and json files). """ from __future__ import with_statement, division from bandicoot.helper.tools import OrderedDict from bandicoot.core import User, Record, Position from bandicoot.helper.tools import percent_records_missing_location, antennas_missing_locations, warning_str from bandicoot.utils import flatten from datetime import datetime from json import dumps from collections import Counter import csv import os def to_csv(objects, filename, digits=5): """ Export the flatten indicators of one or several users to CSV. Parameters ---------- objects : list List of objects to be exported. filename : string File to export to. digits : int Precision of floats. Examples -------- This function can be use to export the results of :meth`bandicoot.utils.all`. >>> U_1 = bc.User() >>> U_2 = bc.User() >>> bc.to_csv([bc.utils.all(U_1), bc.utils.all(U_2)], 'results_1_2.csv') If you only have one object, you can simply pass it as argument: >>> bc.to_csv(bc.utils.all(U_1), 'results_1.csv') """ if not isinstance(objects, list): objects = [objects] data = [flatten(obj) for obj in objects] all_keys = [d for datum in data for d in datum.keys()] field_names = sorted(set(all_keys), key=lambda x: all_keys.index(x)) with open(filename, 'wb') as f: w = csv.writer(f) w.writerow(field_names) def make_repr(item): if item is None: return None elif isinstance(item, float): return repr(round(item, digits)) else: return str(item) for row in data: row = dict((k, make_repr(v)) for k, v in row.items()) w.writerow([make_repr(row.get(k, None)) for k in field_names]) print "Successfully exported %d object(s) to %s" % (len(objects), filename) def to_json(objects, filename): """ Export the indicators of one or several users to JSON. Parameters ---------- objects : list List of objects to be exported. filename : string File to export to. Examples -------- This function can be use to export the results of :meth`bandicoot.utils.all`. >>> U_1 = bc.User() >>> U_2 = bc.User() >>> bc.to_json([bc.utils.all(U_1), bc.utils.all(U_2)], 'results_1_2.json') If you only have one object, you can simply pass it as argument: >>> bc.to_json(bc.utils.all(U_1), 'results_1.json') """ if not isinstance(objects, list): objects = [objects] obj_dict = {obj['name']: obj for obj in objects} with open(filename, 'wb') as f: f.write(dumps(obj_dict, indent=4, separators=(',', ': '))) print "Successfully exported %d object(s) to %s" % (len(objects), filename) def _tryto(function, argument): try: return function(argument) except Exception as ex: return ex def _parse_record(data): def _map_duration(s): return int(s) if s != '' else None def _map_position(data): antenna = Position() if 'antenna_id' in data: antenna.antenna = data['antenna_id'] return antenna elif 'place_id' in data: raise NameError("Use field name 'antenna_id' in input files. 'place_id' is deprecated.") if 'latitude' in data and 'longitude' in data: antenna.position = float(data['latitude']), float(data['longitude']) return antenna return Record(interaction=data['interaction'], direction=data['direction'], correspondent_id=data['correspondent_id'], datetime=_tryto(lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S"), data['datetime']), call_duration=_tryto(_map_duration, data['call_duration']), position=_tryto(_map_position, data)) def filter_record(records): """ Filter records and remove items with missing or inconsistent fields Parameters ---------- records : list A list of Record objects Returns ------- records, ignored : (Record list, dict) A tuple of filtered records, and a dictionary counting the missings fields """ scheme = { 'interaction': lambda r: r.interaction in ['call', 'text'], 'direction': lambda r: r.direction in ['in', 'out'], 'correspondent_id': lambda r: r.correspondent_id is not None, 'datetime': lambda r: isinstance(r.datetime, datetime), 'call_duration': lambda r: isinstance(r.call_duration, (int, float)) if r.interaction == 'call' else True, } ignored = OrderedDict([ ('all', 0), ('interaction', 0), ('direction', 0), ('correspondent_id', 0), ('datetime', 0), ('call_duration', 0), ]) bad_records = [] def _filter(records): global removed for r in records: valid = True for key, test in scheme.iteritems(): if not test(r): ignored[key] += 1 bad_records.append(r) valid = False # Not breaking, we count all fields with errors if valid is True: yield r else: ignored['all'] += 1 return list(_filter(records)), ignored, bad_records def load(name, records, antennas, attributes=None, antennas_path=None, attributes_path=None, describe=False, warnings=False): """ Creates a new user. This function is used by read_csv, read_orange, and read_telenor. If you want to implement your own reader function, we advise you to use the load() function `load` will output warnings on the standard output if some records or antennas are missing a position. Parameters ---------- name : str The name of the user. It is stored in User.name and is useful when exporting metrics about multiple users. records: list A list or a generator of Record objects. antennas : dict A dictionary of the position for each antenna. attributes : dict A (key,value) dictionary of attributes for the current user describe : boolean If describe is True, it will print a description of the loaded user to the standard output. Defaults to false. warnings : boolean, default True If warnings is equal to False, the function will not output the warnings on the standard output. For instance: .. code-block:: python >>> records = [Record(...),...] >>> antennas = {'A51': (37.245265, 115.803418),...} >>> attributes = {'age': 60} >>> load("Frodo", records, antennas, attributes) will returns a new User object. """ user = User() user.name = name user.antennas_path = antennas_path user.attributes_path = attributes_path user.records, ignored, bad_records = filter_record(records) if ignored['all'] != 0: if warnings: print warning_str("Warning: %d record(s) were removed due to missing or incomplete fields." % ignored['all']) for k in ignored.keys(): if k != 'all' and ignored[k] != 0 and warnings: print warning_str(" " * 9 + "%s: %i record(s) with incomplete values" % (k, ignored[k])) user.ignored_records = dict(ignored) if antennas is not None: user.antennas = antennas if attributes is not None: user.attributes = attributes percent_missing = percent_records_missing_location(user) if percent_missing > 0 and warnings: print warning_str("Warning: {0:.2%} of the records are missing a location.".format(percent_missing)) if antennas is None: print warning_str(" No antennas file was given and records are using antennas for position.") if antennas_missing_locations(user) > 0 and warnings: print warning_str("Warning: %d antenna(s) are missing a location." % antennas_missing_locations(user)) sorted_min_records = sorted(set(user.records), key=lambda r: r.datetime) num_dup = len(user.records) - len(sorted_min_records) if num_dup > 0 and warnings: print warning_str("Warning: {0:d} duplicated record(s) were removed.".format(num_dup)) user.records = sorted_min_records if describe is True: user.describe() return user, bad_records def _read_network(user, records_path, attributes_path, read_function, antennas_path=None, extension=".csv"): connections = {} correspondents = Counter([record.correspondent_id for record in user.records]) # Try to load all the possible correspondent files for c_id, count in sorted(correspondents.items()): correspondent_file = os.path.join(records_path, c_id + extension) if os.path.exists(correspondent_file): connections[c_id] = read_function(c_id, records_path, antennas_path, attributes_path, describe=False, network=False, warnings=False) else: connections[c_id] = None def _is_consistent(record): if record.correspondent_id == user.name: correspondent = user elif record.correspondent_id in connections: correspondent = connections[record.correspondent_id] else: return True # consistent by default return True if correspondent is None else record.has_match(correspondent.records) def all_user_iter(): if user.name not in connections: yield user for u in connections.values(): if u is not None: yield u # Filter records and count total number of records before/after num_total_records = sum(len(u.records) for u in all_user_iter()) for u in all_user_iter(): u.records = filter(_is_consistent, u.records) num_total_records_filtered = sum(len(u.records) for u in all_user_iter()) # Report non reciprocated records num_inconsistent_records = num_total_records - num_total_records_filtered if num_inconsistent_records > 0: percent_inconsistent = num_inconsistent_records / num_total_records print warning_str('Warning: {} records ({:.2%}) for all users in the network were not reciprocated. They have been removed.'.format(num_inconsistent_records, percent_inconsistent)) # Return the network dictionary sorted by key return OrderedDict(sorted(connections.items(), key=lambda t: t[0])) def read_csv(user_id, records_path, antennas_path=None, attributes_path=None, network=False, describe=True, warnings=True, errors=False): """ Load user records from a CSV file. Parameters ---------- user_id : str ID of the user (filename) records_path : str Path of the directory all the user files. antennas_path : str, optional Path of the CSV file containing (place_id, latitude, longitude) values. This allows antennas to be mapped to their locations. attributes_path : str, optional Path of the directory containing attributes files (``key, value`` CSV file). Attributes can for instance be variables such as like, age, or gender. Attributes can be helpful to compute specific metrics. network : bool, optional If network is True, bandicoot loads the network of the user's correspondants from the same path. Defaults to False. describe : boolean If describe is True, it will print a description of the loaded user to the standard output. errors : boolean If errors is True, returns a tuple (user, errors), where user is the user object and errors are the records which could not be loaded. Examples -------- >>> user = bandicoot.read_csv('sample_records', '.') >>> print len(user.records) 10 >>> user = bandicoot.read_csv('sample_records', 'samples', sample_places.csv') >>> print len(user.antennas) 5 >>> user = bandicoot.read_csv('sample_records', '.', None, 'sample_attributes.csv') >>> print user.attributes['age'] 25 Notes ----- - The csv files can be single, or double quoted if needed. - Empty cells are filled with ``None``. For example, if the column ``call_duration`` is empty for one record, its value will be ``None``. Other values such as ``"N/A"``, ``"None"``, ``"null"`` will be considered as a text. """ antennas = None if antennas_path is not None: with open(antennas_path, 'rb') as csv_file: reader = csv.DictReader(csv_file) antennas = dict((d['place_id'], (float(d['latitude']), float(d['longitude']))) for d in reader) user_records = os.path.join(records_path, user_id + '.csv') with open(user_records, 'rb') as csv_file: reader = csv.DictReader(csv_file) records = map(_parse_record, reader) attributes = None if attributes_path is not None: user_attributes = os.path.join(attributes_path, user_id + '.csv') try: with open(user_attributes, 'rb') as csv_file: reader = csv.DictReader(csv_file) attributes = dict((d['key'], d['value']) for d in reader) except IOError: attributes = None user, bad_records = load(user_id, records, antennas, attributes, antennas_path, attributes_path=attributes_path, describe=False, warnings=warnings) # Loads the network if network is True: user.network = _read_network(user, records_path, attributes_path, read_csv, antennas_path) user.recompute_missing_neighbors() if describe: user.describe() if errors: return user, bad_records return user def read_orange(user_id, records_path, antennas_path=None, attributes_path=None, network=False, describe=True, warnings=True, errors=False): """ Load user records from a CSV file in *orange* format: ``call_record_type;basic_service;user_msisdn;call_partner_identity;datetime;call_duration;longitude;latitude`` ``basic_service`` takes one of the following values: - 11: telephony; - 12: emergency calls; - 21: short message (in) - 22: short message (out) Parameters ---------- user_id : str ID of the user (filename) records_path : str Path of the directory all the user files. antennas_path : str, optional Path of the CSV file containing (place_id, latitude, longitude) values. This allows antennas to be mapped to their locations. attributes_path : str, optional Path of the directory containing attributes files (``key, value`` CSV file). Attributes can for instance be variables such as like, age, or gender. Attributes can be helpful to compute specific metrics. network : bool, optional If network is True, bandicoot loads the network of the user's correspondants from the same path. Defaults to False. describe : boolean If describe is True, it will print a description of the loaded user to the standard output. errors : boolean If errors is True, returns a tuple (user, errors), where user is the user object and errors are the records which could not be loaded. """ def _parse(reader): records = [] antennas = dict() for row in reader: direction = 'out' if row['call_record_type'] == '1' else 'in' interaction = 'call' if row['basic_service'] in ['11', '12'] else 'text' contact = row['call_partner_identity'] date = datetime.strptime(row['datetime'], "%Y-%m-%d %H:%M:%S") call_duration = float(row['call_duration']) if row['call_duration'] != "" else None lon, lat = float(row['longitude']), float(row['latitude']) latlon = (lat, lon) antenna = None for key, value in antennas.items(): if latlon == value: antenna = key break if antenna is None: antenna = len(antennas) + 1 antennas[antenna] = latlon position = Position(antenna=antenna, location=latlon) record = Record(direction=direction, interaction=interaction, correspondent_id=contact, call_duration=call_duration, datetime=date, position=position) records.append(record) return records, antennas user_records = os.path.join(records_path, user_id + ".csv") fields = ['call_record_type', 'basic_service', 'user_msisdn', 'call_partner_identity', 'datetime', 'call_duration', 'longitude', 'latitude'] with open(user_records, 'rb') as f: reader = csv.DictReader(f, delimiter=";", fieldnames=fields) records, antennas = _parse(reader) attributes = None try: if attributes_path is not None: attributes_file = os.path.join(attributes_path, user_id + ".csv") with open(attributes_file, 'rb') as f: reader = csv.DictReader(f, delimiter=";", fieldnames=["key", "value"]) attributes = {a["key"]: a["value"] for a in reader} except IOError: pass user, bad_records = load(user_id, records, antennas, attributes_path=attributes_path, attributes=attributes, warnings=None, describe=False) if network is True: user.network = _read_network(user, records_path, attributes_path, read_orange, antennas_path) user.recompute_missing_neighbors() if describe: user.describe() if errors: return user, bad_records return user def read_telenor(incoming_cdr, outgoing_cdr, cell_towers, describe=True, warnings=True): """ Load user records from a CSV file in *telenor* format, which is only applicable for call records. .. note:: read_telenor has been deprecated in bandicoot 0.4. Arguments --------- incoming_cdr: str Path to the CSV file containing incoming records, using the following scheme: :: B_PARTY,A_PARTY,DURATION,B_CELL,CALL_DATE,CALL_TIME,CALL_TYPE outgoing_cdr: str Path to the CSV file containing outgoing records, using the following scheme: :: A_NUMBER,B_NUMBER,DURATION,B_CELL,CALL_DATE,CALL_TIME,CALL_TYPE cell_towers: str Path to the CSV file containing the positions of all describe : boolean If describe is True, it will print a description of the loaded user to the standard output. """ print warning_str("read_telenor has been deprecated in bandicoot 0.4.") import itertools import csv def parse_direction(code): if code == 'MOC': return 'out' elif code == 'MTC': return 'in' else: raise NotImplementedError cells = None with open(cell_towers, 'rb') as f: cell_towers_list = csv.DictReader(f) cells = {} for line in cell_towers_list: if line['LONGITUDE'] != '' and line['LATITUDE'] != '': latlon = (float(line['LONGITUDE']), float(line['LATITUDE'])) cell_id = line['CELLID_HEX'] cells[cell_id] = latlon def parse_record(raw): direction = parse_direction(raw['CALL_TYPE'].strip()) if direction == 'in': contact = raw.get('A_PARTY', raw.get('A_NUMBER')) cell_id = raw['B_CELL'] else: contact = raw.get('B_PARTY', raw.get('B_NUMBER')) cell_id = raw['A_CELL'] position = Position(antenna=cell_id, location=cells.get(cell_id)) _date_str = raw.get('CDATE', raw.get('CALL_DATE')) _time_str = raw.get('CTIME', raw.get('CALL_TIME')) _datetime = datetime.strptime(_date_str + _time_str, "%Y%m%d%H:%M:%S") r = Record(interaction='call', direction=direction, correspondent_id=contact, call_duration=float(raw['DURATION'].strip()), datetime=_datetime, position=position) return r with open(incoming_cdr, 'rb') as f_in: incoming_ = map(parse_record, csv.DictReader(f_in)) with open(outgoing_cdr, 'rb') as f: outgoing_ = map(parse_record, csv.DictReader(f)) records = itertools.chain(incoming_, outgoing_) name = incoming_cdr user, errors = load(name, records, cells, warnings=None, describe=False) if describe: user.describe() return user bandicoot-0.4.0/bandicoot/network.py0000755000076500000240000002270012545050672020074 0ustar cynddlstaff00000000000000from __future__ import division from collections import Counter, defaultdict from itertools import groupby, combinations from functools import partial from datetime import datetime, timedelta from bandicoot.utils import all def _round_half_hour(record): """ Round a time DOWN to half nearest half-hour. """ k = record.datetime + timedelta(minutes=-(record.datetime.minute % 30)) return datetime(k.year, k.month, k.day, k.hour, k.minute, 0) def _count_interaction(user, interaction=None, direction='out'): if interaction is 'call_duration': d = defaultdict(int) for r in user.records: if r.direction == direction and r.interaction == 'call': d[r.correspondent_id] += r.call_duration return d if interaction is None: keyfn = lambda x: x.correspondent_id records = (r for r in user.records if r.direction == direction) chunks = groupby(sorted(records, key=keyfn), key=keyfn) # Count the number of distinct half-hour blocks for each user return Counter({c_id: len(set((_round_half_hour(i) for i in items))) for c_id, items in chunks}) if interaction in ['call', 'text']: filtered = [x.correspondent_id for x in user.records if x.interaction == interaction and x.direction == direction] else: raise ValueError("{} is not a correct value of interaction, only 'call'" ", 'text', and 'call_duration' are accepted".format(interaction)) return Counter(filtered) def _interaction_matrix(user, interaction=None, default=0, missing=None): generating_fn = partial(_count_interaction, interaction=interaction) # Just in case, we remove the user from user.network (self records can happen) neighbors = matrix_index(user) def make_direction(direction): rows = [] for u in neighbors: correspondent = user.network.get(u, user) if correspondent is None: # We assume that missing users don't interact with themselves row = [missing if v != u else 0 for v in neighbors] else: cur_out = generating_fn(correspondent, direction=direction) row = [cur_out.get(v, default) for v in neighbors] rows.append(row) return rows m1 = make_direction('out') m2 = make_direction('in') m = [[m1[i][j] if m1[i][j] is not None else m2[j][i] for j in range(len(neighbors))] for i in range(len(neighbors))] return m def matrix_index(user): """ Returns the keys associated with each axis of the matrices. The first key is always the name of the current user, followed by the sorted names of all the correspondants. """ return [user.name] + sorted([k for k in user.network.keys() if k != user.name]) def matrix_directed_weighted(user, interaction=None): """ Returns a directed, weighted matrix for call, text and call duration. If interaction is None, the weight measures both calls and texts: the weight is the number of 30 minutes periods with at least one call or one text. Example ------- >>> m = bc.network.matrix_directed_weighted(user, interaction='call') ``m[i][j]`` is the number of calls from ``i`` to ``j``. """ return _interaction_matrix(user, interaction=interaction) def matrix_directed_unweighted(user): """ Returns a directed, unweighted matrix where an edge exists if there is at least one call or text. """ matrix = _interaction_matrix(user, interaction=None) for a in range(len(matrix)): for b in range(len(matrix)): if matrix[a][b] is not None and matrix[a][b] > 0: matrix[a][b] = 1 return matrix def matrix_undirected_weighted(user, interaction=None): """ Returns an undirected, weighted matrix for call, text and call duration where an edge exists if the relationship is reciprocated. """ matrix = _interaction_matrix(user, interaction=interaction) result = [[0 for _ in range(len(matrix))] for _ in range(len(matrix))] for a in range(len(matrix)): for b in range(len(matrix)): if a != b and matrix[a][b] and matrix[b][a]: result[a][b] = matrix[a][b] + matrix[b][a] elif matrix[a][b] is None or matrix[b][a] is None: result[a][b] = None else: result[a][b] = 0 return result def matrix_undirected_unweighted(user): """ Returns an undirected, unweighted matrix where an edge exists if the relationship is reciprocated. """ matrix = matrix_undirected_weighted(user, interaction=None) for a, b in combinations(range(len(matrix)), 2): if matrix[a][b] > 0 and matrix[b][a] > 0: matrix[a][b], matrix[b][a] = 1, 1 return matrix def clustering_coefficient_unweighted(user): """ The clustering coefficient of the user in the unweighted, undirected ego network. It is defined by counting the number of closed triplets including the current user: .. math:: C = \\frac{2 * \\text{closed triplets}}{ \\text{degree} \, (\\text{degree - 1})} where ``degree`` is the degree of the current user in the network. """ matrix = matrix_undirected_unweighted(user) closed_triplets = 0 for a, b in combinations(xrange(len(matrix)), 2): a_b, a_c, b_c = matrix[a][b], matrix[a][0], matrix[b][0] if a_b is not None and a_c is not None and b_c is not None: if a_b > 0 and a_c > 0 and b_c > 0: closed_triplets += 1. d_ego = sum(matrix[0]) return 2 * closed_triplets / (d_ego * (d_ego - 1)) if d_ego > 1 else 0 def clustering_coefficient_weighted(user, interaction=None): """ The clustering coefficient of the user's weighted, undirected network. It is defined the same way as :meth`~bandicoot.network.clustering_coefficient_unweighted`, except that closed triplets are weighted by the number of interactions. For each triplet (A, B, C), we compute the geometric mean of the number of interactions, using the undirected weighted matrix: .. math:: weight_{abc} = (m_{ab} \; m_{bc} \; m_{ac})^{1/3} The weight is normalized, between 0 and 1, by the maximum value in the matrix. """ matrix = matrix_undirected_weighted(user, interaction=interaction) triplet_weight = 0 max_weight = max(weight for g in matrix for weight in g) for a, b in combinations(range(len(matrix)), 2): a_b, a_c, b_c = matrix[a][b], matrix[a][0], matrix[b][0] if a_b is not None and a_c is not None and b_c is not None: if a_b and a_c and b_c: triplet_weight += (a_b * a_c * b_c) ** (1 / 3) / max_weight d_ego = sum(1 for i in matrix[0] if i > 0) return 2 * triplet_weight / (d_ego * (d_ego - 1)) if d_ego > 1 else 0 def assortativity_indicators(user): """ Computes the assortativity of indicators. This indicator measures the similarity of the current user with his correspondants, for all bandicoot indicators. For each one, it calculates the variance of the current user's value with the values for all his correspondants: .. math:: \\text{assortativity}(J) = \\frac{1}{n} \\sum_i^n (J_{\\text{user}} - J_{\\text{i}})^2 for the indicator :math:`J`, and all the :math:`n` correspondents. """ matrix = matrix_undirected_unweighted(user) count_indicator = defaultdict(int) total_indicator = defaultdict(int) # Use all indicator except reporting variables and attributes ego_indics = all(user, flatten=True) ego_indics = {a: value for a, value in ego_indics.items() if a != "name" and a[:11] != "reporting__" and a[:10] != "attributes"} for i, u_name in enumerate(matrix_index(user)): correspondent = user.network.get(u_name, None) if correspondent is None or u_name == user.name or matrix[0][i] == 0: # Non reciprocated edge continue neighbor_indics = all(correspondent, flatten=True) for a in ego_indics: if ego_indics[a] is not None and neighbor_indics[a] is not None: total_indicator[a] += 1 count_indicator[a] += (ego_indics[a] - neighbor_indics[a]) ** 2 assortativity = {} for i in count_indicator: assortativity[i] = count_indicator[i] / total_indicator[i] return assortativity def assortativity_attributes(user): """ Computes the assortativity of the nominal attributes. This indicator measures the homophily of the current user with his correspondants, for each attributes. It returns a value between 0 (no assortativity) and 1 (all the contacts share the same value): the percentage of contacts sharing the same value. """ matrix = matrix_undirected_unweighted(user) neighbors = [k for k in user.network.keys() if k != user.name] neighbors_attrbs = {} for i, u_name in enumerate(matrix_index(user)): correspondent = user.network.get(u_name, None) if correspondent is None or u_name == user.name or matrix[0][i] == 0: continue if correspondent.has_attributes: neighbors_attrbs[correspondent.name] = correspondent.attributes assortativity = {} for a in user.attributes: total = sum(1 for n in neighbors if n in neighbors_attrbs and user.attributes[a] == neighbors_attrbs[n][a]) den = sum(1 for n in neighbors if n in neighbors_attrbs) assortativity[a] = total / den if den != 0 else None return assortativity bandicoot-0.4.0/bandicoot/spatial.py0000755000076500000240000001045612545050672020045 0ustar cynddlstaff00000000000000from __future__ import division import math from .helper.group import spatial_grouping, group_records, statistics, _binning from .helper.tools import entropy, great_circle_distance, pairwise from collections import Counter from collections import defaultdict @spatial_grouping(user_kwd=True) def percent_at_home(positions, user): """ The percentage of interactions the user had while he was at home. Notes ----- The position of the home is computed by :meth:`User.compute_home`. If no home can be found, the percentage at home will be ``None``. """ if not user.has_home: return None if user.home.antenna is None and user.home.location is None: return None positions = list(positions) total_home = sum(1 for p in positions if p == user.home) return float(total_home) / len(positions) if len(positions) != 0 else 0 @spatial_grouping def radius_of_movement(records): return NotImplemented @spatial_grouping(user_kwd=True) def radius_of_gyration(positions, user): """ The radius of gyration. The radius of gyration is the *equivalent distance* of the mass from the center of gravity, for all visited places [GON2008]_ None is returned if there are no (lat, lon) positions for where the user has been. .. [GON2008] Gonzalez, M. C., Hidalgo, C. A., & Barabasi, A. L. (2008). Understanding individual human mobility patterns. Nature, 453(7196), 779-782. """ d = Counter(p._get_location(user) for p in positions if p._get_location(user) is not None) sum_weights = sum(d.values()) positions = d.keys() # Unique positions if len(positions) == 0: return None barycenter = [0, 0] for pos, t in d.items(): barycenter[0] += pos[0] * t barycenter[1] += pos[1] * t barycenter[0] /= sum_weights barycenter[1] /= sum_weights r = 0. for pos, t in d.items(): r += float(t) / sum_weights * great_circle_distance(barycenter, pos) ** 2 r += float(t) / sum_weights * great_circle_distance(barycenter, pos) ** 2 return math.sqrt(r) @spatial_grouping def entropy_of_antennas(positions, normalize=False): """ The entropy of visited antennas. Parameters ---------- normalize: boolean, default is False Returns a normalized entropy between 0 and 1. """ counter = Counter(p for p in positions) raw_entropy = entropy(counter.values()) n = len(counter) if normalize and n > 1: return raw_entropy / math.log(n) else: return raw_entropy @spatial_grouping def number_of_antennas(positions): """ The number of unique places visited. """ return len(set(positions)) @spatial_grouping def frequent_antennas(positions, percentage=0.8): """ The number of location that account for 80% of the locations where the user was. Percentage can be supplied as a decimal (e.g., .8 for default 80%). """ location_count = Counter(map(str, positions)) target = math.ceil(sum(location_count.values()) * percentage) location_sort = sorted(location_count.keys(), key=lambda x: location_count[x]) while target > 0 and len(location_sort) > 0: location_id = location_sort.pop() target -= location_count[location_id] return len(location_count) - len(location_sort) def churn_rate(user, summary='default', **kwargs): """ The cosine distance between the frequency spent at each tower each week. """ if len(user.records) == 0: return None iter = group_records(user, groupby='week') weekly_positions = [list(_binning(l)) for l in iter] # bin positions every 30 minutes all_positions = list(set(p for l in weekly_positions for p in l)) frequencies = {} cos_dist = [] for week, week_positions in enumerate(weekly_positions): count = Counter(week_positions) total = sum(count.values()) frequencies[week] = [count.get(p, 0) / total for p in all_positions] all_indexes = xrange(len(all_positions)) for f_1, f_2 in pairwise(frequencies.values()): num = sum(f_1[a] * f_2[a] for a in all_indexes) denom_1 = sum(f ** 2 for f in f_1) denom_2 = sum(f ** 2 for f in f_2) cos_dist.append(1 - num / (denom_1 ** .5 * denom_2 ** .5)) return statistics(cos_dist, summary=summary) bandicoot-0.4.0/bandicoot/special/0000755000076500000240000000000012545051250017436 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/special/__init__.py0000644000076500000240000000007612545050666021564 0ustar cynddlstaff00000000000000__all__ = ['punchcard', 'demo'] from . import punchcard, demobandicoot-0.4.0/bandicoot/special/demo.py0000644000076500000240000000523512545050666020753 0ustar cynddlstaff00000000000000import os import bandicoot as bc from collections import Counter def export_antennas(user, directory): with open(os.path.join(directory, 'antenna.csv'), 'wb') as f: f.write('id,latitude,longitude,interactions\n') for id, (lat, lon) in user.antennas.items(): interactions = sum([1 for r in user.records if r.position.antenna == id]) f.write("{},{},{},{}\n".format(id, lat, lon, interactions)) def export_transitions(user, directory): antennas_id = [r.position.antenna for r in user.records] transitions = Counter([(min(i, j), max(i, j)) for (i, j) in zip(antennas_id, antennas_id[1:])]) with open(os.path.join(directory, 'transitions.csv'), 'wb') as f: f.write('source,target,amount\n') for (source, target), count in transitions.items(): if source != target: f.write("{},{},{}\n".format(source, target, count)) def export_timeline(user, directory): with open(os.path.join(directory, 'timeseries.csv'), 'wb') as f: f.write('time,type,call_duration\n') for r in user.records: time = r.datetime.strftime('%d-%m-%y %H:%M') interaction = ('inc_' if r.direction == 'in' else 'out_') + r.interaction call_duration = r.call_duration or '' f.write("{},{},{}\n".format(time, interaction, call_duration)) def export_network(user, directory): m_texts = bc.network.matrix_directed_weighted(user, 'text') m_calls = bc.network.matrix_directed_weighted(user, 'call') nb_users = len(m_texts) links = [] for i in range(nb_users): for j in range(nb_users): source_calls = m_calls[j][i] or 0 target_calls = m_calls[i][j] or 0 source_texts = m_texts[j][i] or 0 target_texts = m_texts[i][j] or 0 if i < j and source_calls + target_calls + source_texts + target_texts > 0: links.append([i, j, (source_calls, target_calls, source_texts, target_texts)]) names = [user.name] + [i for i in user.network.keys() if i != user.name] nodes = set(l[0] for l in links) | set(l[1] for l in links) with open(os.path.join(directory, 'nodes.csv'), 'wb') as f: f.write('name,no_network_info\n') for n in nodes: f.write("{},{}\n".format(names[n], int(None in m_texts[n] or None in m_calls[n]))) compress_ids = dict(zip(nodes, range(len(nodes)))) with open(os.path.join(directory, 'links.csv'), 'wb') as f: f.write('source,target,source_calls,target_calls,source_texts,target_texts\n') for i, j, data in links: f.write("{},{},{},{},{},{}\n".format(*(compress_ids[i], compress_ids[j]) + data)) bandicoot-0.4.0/bandicoot/special/punchcard.py0000644000076500000240000002524112545050666021775 0ustar cynddlstaff00000000000000import bandicoot as bc from bandicoot.helper.group import group_records from bandicoot.core import User, Record from bisect import bisect_right from functools import partial import datetime as dt import itertools import csv import math def create_punchcards(user, split_interval=60): """ Computes raw indicators (e.g. number of outgoing calls) for intervals of ~1 hour across each week of user data. These "punchcards" are returned in a nested list with each sublist containing [user.name, channel, weekday, section, value]. Parameters ---------- user : object The user to create punchcards for. split_interval : int The interval in minutes for which each indicator is computed. Defaults to 60. Needs to be able to split a day (24*60 minutes) evenly. """ if not float(24 * 60 / split_interval).is_integer(): raise ValueError( "The minute interval set for the punchcard structure does not evenly divide the day!") contacts_in = partial(bc.individual.number_of_contacts, direction='in', interaction='callandtext', summary=None) contacts_out = partial(bc.individual.number_of_contacts, direction='out', interaction='callandtext', summary=None) calls_in = partial(bc.individual.number_of_interactions, direction='in', interaction='call', summary=None) calls_out = partial(bc.individual.number_of_interactions, direction='out', interaction='call', summary=None) texts_in = partial(bc.individual.number_of_interactions, direction='in', interaction='text', summary=None) texts_out = partial(bc.individual.number_of_interactions, direction='out', interaction='text', summary=None) time_spent_in = partial(bc.individual.call_duration, direction='in', interaction='call', summary=None) time_spent_out = partial(bc.individual.call_duration, direction='out', interaction='call', summary=None) core_func = [ (contacts_in, "scalar"), (contacts_out, "scalar"), (calls_in, "scalar"), (calls_out, "scalar"), (texts_in, "scalar"), (texts_out, "scalar") ] time_func = [ (time_spent_in, "summarystats"), (time_spent_out, "summarystats") ] pc = [] sections = [ (i + 1) * split_interval for i in range(7 * 24 * 60 / split_interval)] temp_user = _extract_user_info(user) for grouped_records in group_records(user, groupby='week'): week_records = list(grouped_records) time_spent_rec = _transform_to_time_spent( week_records, split_interval, sections) pc.extend(_calculate_channels( week_records, sections, split_interval, core_func, temp_user)) pc.extend(_calculate_channels( time_spent_rec, sections, split_interval, time_func, temp_user, len(core_func))) return pc def to_csv(punchcards, filename, digits=5): """ Exports a list of punchcards to a specified filename in the CSV format. Parameters ---------- punchcards : list The punchcards to export. filename : string Path for the exported CSV file. """ with open(filename, 'wb') as f: w = csv.writer(f) w.writerow(['year_week', 'channel', 'weekday', 'section', 'value']) def make_repr(item): if item is None: return None elif isinstance(item, float): return repr(round(item, digits)) else: return str(item) for row in punchcards: w.writerow([make_repr(item) for item in row]) def read_csv(filename): """ Read a list of punchcards from a CSV file. """ with open(filename, 'rb') as f: r = csv.reader(f) next(r) # remove header pc = list(r) # remove header and convert to numeric for i, row in enumerate(pc): row[1:4] = map(int, row[1:4]) row[4] = float(row[4]) return pc def _calculate_channels(records, sections, split_interval, channel_funcs, user, c_start=0): """ Used to group a list of records across a week as defined by the supplied sections. Outputs a list containing records in each section and a list with info to identify those sections. Parameters ---------- records : list The week of records to calculate the channels for. sections : list The list of sections for grouping. Each section will have an integer value stating the minutes away from midnight between Sunday and Monday. split_interval : int The interval in minutes for which each indicator is computed. channel_funcs : list Indicator functions that generate the values for the punchcard. user : object The user to calculate channels for. c_start : num Start numbering of channels from this value. Optional parameter. Default value of 0. Used when adding channels to the same user using different lists of records. """ week_matrix = [] if len(records) == 0: return week_matrix if not isinstance(records, list): records = [records] year_week = str(records[0].datetime.isocalendar()[ 0]) + '-' + str(records[0].datetime.isocalendar()[1]) section_lists, section_id = _punchcard_grouping(records, sections, split_interval) for c, fun in enumerate(channel_funcs): for b, section_records in enumerate(section_lists): indicator_fun, return_type = fun # _records is used to avoid recomputing home user._records = section_records output = indicator_fun(user)['allweek']['allday'].values()[0] if return_type == 'scalar': indicator = sum(d for d in output if d is not None) elif return_type == 'summarystats': indicator = sum(d for group in output for d in group if d is not None) if indicator != 0: week_matrix.append( [year_week, c + c_start, section_id[b][0], section_id[b][1], float(indicator)]) return week_matrix def _punchcard_grouping(records, sections, split_interval): """ Used to group a list of records across a week as defined by the supplied sections. Outputs a list containing records in each section and a list with info to identify those sections. Parameters ---------- records : list The week of records to group across the different weekdays/sections. sections : list The list of sections for grouping. Each section will have an integer value stating the minutes away from midnight between Sunday and Monday. split_interval : int The interval in minutes for which each indicator is computed. """ def _group_by_weektime(records, sections): for _, group in itertools.groupby(records, key=lambda r: bisect_right(sections, _find_weektime(r.datetime))): yield group section_records = _group_by_weektime(records, sections) section_lists = _extract_list_from_generator(section_records) section_indices = [bisect_right( sections, _find_weektime(r_list[0].datetime)) for r_list in section_lists] section_id = _find_day_section_from_indices( section_indices, split_interval) assert(len(section_lists) == len(section_id) and len( section_indices) == len(set(section_indices))) return section_lists, section_id def _transform_to_time_spent(records, split_interval, sections): """ Each call that crosses a boundary of the sections in the punchcard is split. These new records contain the amount of time (in record.call_duration) spent talking in that specific section. """ t_records = [] week_nr = records[0].datetime.isocalendar()[1] # contrary to the rest of the binning process, this is done with second # precision for r in filter(lambda rec: rec.interaction == 'call' and rec.call_duration > 0, records): t_left = r.call_duration t_to_next_section = _seconds_to_section_split(r, sections) t_spent_total = 0 while (t_left > 0): t_spent = min(t_to_next_section, t_left) dt_new = r.datetime + dt.timedelta(seconds=t_spent_total) if dt_new.isocalendar()[1] > week_nr: dt_new -= dt.timedelta(days=7) t_records.append( Record('call', r.direction, None, dt_new, t_spent, None)) t_left -= t_spent t_spent_total += t_spent t_to_next_section = split_interval * 60 return sorted(t_records, key=lambda r: _find_weektime(r.datetime)) def _extract_user_info(user): """ Creates a new user class with extracted user attributes for later use. A new user is needed when wanting to avoid overwritting e.g. ``user.records``. """ temp_user = User() copy_attributes = [ 'antennas', 'name', 'night_start', 'night_end', 'weekend', 'home'] for attr in copy_attributes: setattr(temp_user, attr, getattr(user, attr)) return temp_user def _find_weektime(datetime, time_type='min'): """ Finds the minutes/seconds aways from midnight between Sunday and Monday. Parameters ---------- datetime : datetime The date and time that needs to be converted. time_type : 'min' or 'sec' States whether the time difference should be specified in seconds or minutes. """ if time_type == 'sec': return datetime.weekday() * 24 * 60 * 60 + datetime.hour * 60 * 60 + datetime.minute * 60 + datetime.second elif time_type == 'min': return datetime.weekday() * 24 * 60 + datetime.hour * 60 + datetime.minute else: raise ValueError("Invalid time type specified.") def _extract_list_from_generator(generator): """ Iterates over a generator to extract all the objects and add them to a list. Useful when the objects have to be used multiple times. """ extracted = [] for i in generator: extracted.append(list(i)) return extracted def _seconds_to_section_split(record, sections): """ Finds the seconds to the next section from the datetime of a record. """ next_section = sections[ bisect_right(sections, _find_weektime(record.datetime))] * 60 return next_section - _find_weektime(record.datetime, time_type='sec') def _find_day_section_from_indices(indices, split_interval): """ Returns a list with [weekday, section] identifiers found using a list of indices. """ cells_day = 24 * 60 / split_interval return [[int(math.floor(i / cells_day)), i % cells_day] for i in indices] bandicoot-0.4.0/bandicoot/tests/0000755000076500000240000000000012545051250017160 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot/tests/__init__.py0000644000076500000240000000004612545050666021303 0ustar cynddlstaff00000000000000from generate_user import sample_user bandicoot-0.4.0/bandicoot/tests/generate_user.py0000755000076500000240000001562112545050666022404 0ustar cynddlstaff00000000000000#!/usr/bin/python import hashlib import datetime import random import math import csv import copy from bandicoot.core import Record, Position import bandicoot as bc from bandicoot.helper.tools import OrderedDict def random_record(**kwargs): n_users = 48 rate = 1e-4 year = 2012 month = random.choice(range(1, 3)) day = random.choice(range(1, 28)) # ensures that some correspondents have more interactions than others correspondent = random.randint( 0, n_users / 2) + random.randint(0, n_users / 2) r = {'datetime': datetime.datetime(year, month, day) + datetime.timedelta(seconds=math.floor(-1 / rate * math.log(random.random()))), 'interaction': random.choice(['text', 'text', 'text', 'call', 'call']), 'correspondent_id': "correspondent_{}".format(correspondent), 'direction': random.choice(['in', 'in', 'out']), 'call_duration': random.randint(1, 1000), 'position': Position(location=(random.uniform(-5, 5), random.uniform(-5, 5)))} if r['interaction'] == "text": r['call_duration'] = None r.update(kwargs) return Record(**r) def sample_user(number_records=1482, seed=42, pct_in_network=0.8): old_state = random.getstate() random.seed(seed) towers = { 701: (42.3555, -71.099541), 702: (42.359039, -71.094595), 703: (42.360481, -71.087321), 704: (42.361013, -71.097868), 705: (42.370849, -71.114613), 706: (42.3667427, -71.1069847), 707: (42.367589, -71.076537) } towers_position = [Position(antenna=k, location=v) for k, v in towers.items()] ego_records = [random_record( position=random.choice(towers_position)) for _ in xrange(number_records)] user, _ = bc.io.load( "sample_user", ego_records, towers, None, describe=False) # create network correspondents = set([record.correspondent_id for record in ego_records]) correspondent_records = {} connections = {} n_in_network = int(len(correspondents) * pct_in_network) if n_in_network % 2 != 0: n_in_network = n_in_network - 1 in_network_correspondents = random.sample(correspondents, n_in_network) def reverse_records(records, current_owner): for r in records: r.direction = 'out' if r.direction == 'in' else 'in' r.correspondent_id = current_owner return records # set records from ego for c_id in sorted(in_network_correspondents): reciprocal_records = filter( lambda r: r.correspondent_id == c_id, ego_records) reciprocal_records = reverse_records( copy.deepcopy(reciprocal_records), "ego") correspondent_records[c_id] = reciprocal_records def generate_group_with_random_links(pct_users_in_group): n_in_group = int(len(correspondents) * pct_users_in_group) group = random.sample(non_grouped_correspondents, n_in_group) networkusers_group = list() for user in group: if user in in_network_correspondents: networkusers_group.append(user) def create_pair(source): user_pair = [source, random.sample(group, 1)[0]] if user_pair[0] in non_grouped_correspondents: non_grouped_correspondents.remove(user_pair[0]) if user_pair[1] in non_grouped_correspondents: non_grouped_correspondents.remove(user_pair[1]) extra_records = [random_record(position=random.choice(towers_position), interaction=random.choice( ['text', 'call', 'call']), correspondent_id=user_pair[1]) for _ in xrange(random.randrange(25, 150))] correspondent_records[user_pair[0]].extend(extra_records) if (user_pair[1] in in_network_correspondents): correspondent_records[user_pair[1]].extend( reverse_records(copy.deepcopy(extra_records), user_pair[0])) # create pairs of users for i in range(len(networkusers_group)): create_pair(networkusers_group[i]) if random.choice(range(2)) == 0: create_pair(networkusers_group[i]) non_grouped_correspondents = copy.deepcopy(correspondents) for i in range(4): generate_group_with_random_links(pct_users_in_group=0.4 - i * 0.1) # create user object for c_id in sorted(correspondents): if (c_id in in_network_correspondents): correspondent_user, _ = bc.io.load( c_id, correspondent_records[c_id], towers, None, describe=False) else: correspondent_user = None connections[c_id] = correspondent_user # return the network dictionary sorted by key user.network = OrderedDict(sorted(connections.items(), key=lambda t: t[0])) user.recompute_missing_neighbors() random.setstate(old_state) return user def write_new_user(filepath, n=1960): user = sample_user(n) schema = ['interaction', 'direction', 'correspondent_id', 'datetime', 'call_duration', 'antenna_id'] with open(filepath, "wb") as new_user: w = csv.writer(new_user) w.writerow(schema) for record in user.records: w.writerow([record.position.antenna if x == 'antenna_id' else getattr(record, x) for x in schema]) print "Finished writing new user to " + filepath write_answers(user) def write_answers(user): bc.io.to_json([bc.utils.all(user, groupby='week', summary='default')], "samples/automatic/automatic_result.json") bc.io.to_json([bc.utils.all(user, groupby='week', summary='extended')], "samples/automatic/automatic_result_extended.json") bc.io.to_json([bc.utils.all(user, groupby=None, summary='default')], "samples/automatic/automatic_result_no_grouping.json") def random_burst(count, delta=datetime.timedelta(minutes=10), **kwargs): first_date = datetime.datetime(2014, 01, 01, 10, 41) for i in range(count): _date = first_date + delta * i yield random_record(datetime=_date, **kwargs) def random_records(n, antennas, number_of_users=150, ingoing=0.7, percent_text=0.3, rate=1e-4): current_date = datetime.datetime(2013, 01, 01, 00, 00, 00) results = [] for _ in range(n): current_date += datetime.timedelta( seconds=math.floor(-1 / rate * math.log(random.random()))) interaction = "text" if random.random() < percent_text else "call" r = Record( interaction=interaction, direction='in' if random.random() < ingoing else 'out', correspondent_id=hashlib.md5( str(random.randint(1, number_of_users))).hexdigest(), datetime=str(current_date), call_duration=random.randint( 1, 1000) if interaction == "call" else '', position=random.choice(antennas) ) results.append(r._asdict()) return results bandicoot-0.4.0/bandicoot/tests/test_core.py0000644000076500000240000000361512545050666021540 0ustar cynddlstaff00000000000000""" Tests for bandicoot.core (User, Position, and Record classes) """ import bandicoot as bc import unittest import datetime from testing_tools import parse_dict import os class TestCore(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestCore._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestCore._dir_changed = True self.user = bc.io.read_orange("u_test", "samples", describe=False) self.user_nocturnal = bc.io.read_orange("u_test", "samples", describe=False) self.user_nocturnal.night_start = datetime.time(7, 0) self.user_nocturnal.night_end = datetime.time(19, 0) self.maxDiff = None def test_recompute_home_place(self): old_home = self.user.home new_home = self.user.recompute_home() self.assertEqual(old_home, new_home) self.assertEqual(old_home, self.user.home) def test_set_home(self): towers = parse_dict("samples/towers.json") towers = {key: tuple(value) for (key, value) in towers.items()} new_home = bc.core.Position(antenna=towers["1"]) self.user.set_home(new_home) self.assertEqual(self.user.home, new_home) new_home = bc.core.Position(location=(42.3555368, -71.099507)) self.user.set_home((42.3555368, -71.099507)) self.assertEqual(self.user.home, new_home) class TestTowers(unittest.TestCase): def setUp(self): self.user = bc.io.read_csv("A", "samples/manual/", "samples/towers.csv", describe=False) def testParse(self): towers = parse_dict("samples/towers.json") towers = {key: tuple(value) for (key, value) in towers.items()} self.assertDictEqual(self.user.antennas, towers) bandicoot-0.4.0/bandicoot/tests/test_export.py0000644000076500000240000000452012545050666022125 0ustar cynddlstaff00000000000000import bandicoot as bc import unittest from testing_tools import parse_dict, file_equality, metric_suite, compare_dict import tempfile import os from collections import OrderedDict as OD class TestExport(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestExport._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestExport._dir_changed = True def test_different_keys(self): tmp_file = tempfile.NamedTemporaryFile() dict1 = {"x": 1, "y": 2, "z": 3} dict1 = OD(sorted(dict1.items(), key=lambda t: t[0])) dict2 = {"a": 4, "b": 5, "c": 6} dict2 = OD(sorted(dict2.items(), key=lambda t: t[0])) bc.io.to_csv([dict1, dict2], tmp_file.name) self.assertTrue(file_equality(tmp_file.name, "samples/to_csv_different_keys.csv")) def test_same_keys(self): tmp_file = tempfile.NamedTemporaryFile() dict1 = {"a": 1, "b": 2, "c": 3} dict1 = OD(sorted(dict1.items(), key=lambda t: t[0])) dict2 = {"a": 4, "b": 5, "c": 6} dict2 = OD(sorted(dict2.items(), key=lambda t: t[0])) bc.io.to_csv([dict1, dict2], tmp_file.name) self.assertTrue(file_equality(tmp_file.name, "samples/to_csv_same_keys.csv")) def test_different_json(self): tmp_file = tempfile.NamedTemporaryFile() dict1 = {"name": "dict1", "x": 1, "y": 2, "z": 3} dict1 = OD(sorted(dict1.items(), key=lambda t: t[0])) dict2 = {"name": "dict2", "a": 4, "b": 5, "c": 6} dict2 = OD(sorted(dict2.items(), key=lambda t: t[0])) bc.io.to_json([dict1, dict2], tmp_file.name) self.assertTrue(file_equality(tmp_file.name, "samples/to_json_different_keys.json")) def test_different_json(self): tmp_file = tempfile.NamedTemporaryFile() dict1 = {"name": "dict1", "a": 1, "b": 2, "c": 3} dict1 = OD(sorted(dict1.items(), key=lambda t: t[0])) dict2 = {"name": "dict2", "a": 4, "b": 5, "c": 6} dict2 = OD(sorted(dict2.items(), key=lambda t: t[0])) bc.io.to_json([dict1, dict2], tmp_file.name) self.assertTrue(file_equality(tmp_file.name, "samples/to_json_same_keys.json")) bandicoot-0.4.0/bandicoot/tests/test_group.py0000644000076500000240000001562712545050666021752 0ustar cynddlstaff00000000000000""" Test for the bandicoot.helper.group module. """ import bandicoot as bc from bandicoot.core import Record, Position import unittest import datetime from bandicoot.tests.generate_user import random_burst from bandicoot.helper.group import group_records from bandicoot.helper.tools import std, mean from datetime import timedelta import numpy as np import os class TestGroup(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestGroup._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestGroup._dir_changed = True self.maxDiff = None self.user = bc.io.read_orange("u_test", "samples", describe=False) self.random_int_list = np.random.randint(1, 1000, size=9001) self.sum_stats_list = [bc.helper.tools.SummaryStats(np.random.rand(), np.random.rand(), np.random.rand(), np.random.rand(), np.random.rand(), np.random.rand(), np.random.rand(), []) for _ in range(9001)] def test_statistics(self): self.assertDictEqual(bc.helper.group.statistics(self.random_int_list, summary='default'), { 'mean': mean(self.random_int_list), 'std': std(self.random_int_list), }) def mean_std(key): return { 'mean': mean([getattr(s, key) for s in self.sum_stats_list]), 'std': std([getattr(s, key) for s in self.sum_stats_list]), } self.assertDictEqual(bc.helper.group.statistics(self.sum_stats_list, summary='extended'), { 'kurtosis': mean_std('kurtosis'), 'mean': mean_std('mean'), 'median': mean_std('median'), 'skewness': mean_std('skewness'), 'std': mean_std('std'), 'min': mean_std('min'), 'max': mean_std('max') }) self.assertEqual(bc.helper.group.statistics([]).values(), [None] * 2) def test_statistics_bad_aggregated(self): def run_bad_aggregated(): try: bc.helper.group.statistics("bad_aggregated") except (TypeError, ValueError): return True return False self.assertTrue(run_bad_aggregated()) def test_weekly_group(self): records = [ Record("test_itr", "in", "1", datetime.datetime(2014, 8, 24), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 9, 4), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 9, 11), 1, Position()) ] user = bc.User() user.records = records grouping = bc.helper.group.group_records(user, groupby='week') groups = [[r for r in l] for l in grouping] self.assertEqual(groups, [[records[0]], [records[1]], [records[2]]]) def test_weekday_group(self): records = [ Record("test_itr", "in", "1", datetime.datetime(2014, 8, 25), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 9, 4), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 9, 11), 1, Position()) ] user = bc.User() user.records = records grouping = bc.helper.group.group_records(user, groupby='week', part_of_week='weekday') groups = [[r for r in l] for l in grouping] self.assertEqual(groups, [[records[0]], [records[1]], [records[2]]]) def test_weekend_group(self): records = [ Record("test_itr", "in", "1", datetime.datetime(2014, 8, 23), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 8, 31), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 10, 18), 1, Position()) ] user = bc.User() user.records = records grouping = bc.helper.group.group_records(user, groupby='week', part_of_week='weekend') groups = [[r for r in l] for l in grouping] self.assertEqual(groups, [[records[0]], [records[1]], [records[2]]]) def test_daily_group(self): records = [ Record("test_itr", "in", "1", datetime.datetime(2014, 8, 22, 10, 00), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 8, 23, 10, 00), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 9, 7, 11, 00), 1, Position()), Record("test_itr", "in", "1", datetime.datetime(2014, 10, 18, 2, 00), 1, Position()) ] user = bc.User() user.records = records grouping = bc.helper.group.group_records(user, groupby='week', part_of_day='night') groups = [[r for r in l] for l in grouping] self.assertEqual(groups, [[records[3]]]) grouping = bc.helper.group.group_records(user, groupby='week', part_of_day='day') groups = [[r for r in l] for l in grouping] self.assertEqual(groups, [[records[0], records[1]], [records[2]]]) def test_none_group(self): records = [ Record("call", "in", "1", datetime.datetime(2014, 9, 4), 1, Position()), Record("call", "in", "1", datetime.datetime(2014, 9, 5), 1, Position()), Record("call", "in", "1", datetime.datetime(2014, 9, 11), 1, Position()), Record("call", "in", "1", datetime.datetime(2014, 9, 12), 1, Position()) ] user = bc.User() user.records = records grouping = bc.helper.group.group_records(user, groupby=None) self.assertEqual(records, list(next(grouping))) self.assertRaises(StopIteration, grouping.next) class ConsistencyTests(unittest.TestCase): def setUp(self): self.user = bc.User() self.user.records = random_burst(100, delta=timedelta(days=2)) def _group_set(self, method, interaction): chunks = group_records(self.user, groupby=method, interaction=interaction) new_records = set(r for c in chunks for r in c) return new_records def test_weekly(self): old_records = set(self.user.records) new_records = self._group_set('week', None) self.assertSetEqual(new_records, old_records) new_records = self._group_set('week', 'call') self.assertSetEqual(new_records, {r for r in old_records if r.interaction == 'call'}) class MissingTests(unittest.TestCase): def setUp(self): self.user = bc.read_csv('user_ignored', 'samples') def test_amount(self): result = { 'all': 4, 'interaction': 2, 'direction': 2, 'correspondent_id': 0, 'datetime': 0, 'call_duration': 1, } self.assertDictEqual(self.user.ignored_records, result) def test_total_records(self): self.assertEqual(len(self.user.records), 1) bandicoot-0.4.0/bandicoot/tests/test_manual.py0000644000076500000240000000252212545050666022061 0ustar cynddlstaff00000000000000""" Test manual user files, made by real humans, with all metrics handcrafted. """ import bandicoot as bc import unittest from testing_tools import parse_dict, metric_suite import os class TestManual(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestManual._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestManual._dir_changed = True self.user_A = bc.io.read_csv("A", "samples/manual", "samples/towers.csv", describe=False, network=False) self.user_B = bc.io.read_csv("B", "samples/manual", "samples/towers.csv", describe=False, network=False) self.user_A_orange = bc.io.read_orange("A_orange", "samples/manual", describe=False, network=False) def test_A_metrics(self): self.assertTrue(*metric_suite(self.user_A, parse_dict("samples/manual/A.json"), groupby=None, decimal=4)) def test_B_metrics(self): self.assertTrue(*metric_suite(self.user_B, parse_dict("samples/manual/B.json"), groupby=None, decimal=3)) def test_A_orange_metrics(self): self.assertTrue(*metric_suite(self.user_A_orange, parse_dict("samples/manual/A.json"), groupby=None, decimal=4))bandicoot-0.4.0/bandicoot/tests/test_network.py0000644000076500000240000000472312545050666022302 0ustar cynddlstaff00000000000000""" Test the unweighted and weighted clustering methods from bandicoot.network. """ import bandicoot as bc import unittest import os import networkx as nx class TestClustering(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestClustering._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestClustering._dir_changed = True self.user = bc.io.read_csv("ego", "samples/network", network=True, describe=False, warnings=False) def test_unweighted_clustering_coefficient(self): G = nx.Graph() G.add_edges_from([ ('ego', 'A'), ('ego', 'B'), ('ego', 'F'), ('A', 'B'), ('B', 'D'), ('F', 'H') ]) bc_clustering_coeff = bc.network.clustering_coefficient_unweighted(self.user) nx_clustering_coeff = nx.clustering(G, 'ego') self.assertAlmostEqual(bc_clustering_coeff, nx_clustering_coeff) def test_weighted_clustering_coefficient(self, interaction=None): G = nx.Graph() G.add_weighted_edges_from([ ('ego', 'A', 4), ('ego', 'B', 4), ('ego', 'F', 3), ('A', 'B', 2), ('F', 'H', 2) ]) bc_clustering_coeff = bc.network.clustering_coefficient_weighted( self.user, interaction=interaction) nx_clustering_coeff = nx.clustering(G, 'ego', weight='weight') self.assertAlmostEqual(bc_clustering_coeff, nx_clustering_coeff) class TestAssortativity(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestAssortativity._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestAssortativity._dir_changed = True self.user = bc.io.read_csv("ego", "samples/network", attributes_path="samples/attributes", network=True, describe=False, warnings=False) def test_attributes_assortativity(self): self.assertEqual(bc.network.assortativity_attributes(self.user), { 'gender': 0.0, 'age': 1.0, 'is_subscriber': 1.0, 'individual_id': 0.0}) bandicoot-0.4.0/bandicoot/tests/test_parsers.py0000644000076500000240000000601112545050666022260 0ustar cynddlstaff00000000000000""" Test the import of CSV files. """ import bandicoot as bc from bandicoot.core import Record, Position import unittest from StringIO import StringIO import sys import datetime import csv import os class TestParsers(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestParsers._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestParsers._dir_changed = True def test_read_orange(self): user = bc.io.read_orange("u_test", "samples", describe=False) self.assertEqual(len(user.records), 500) def test_read_csv(self): user = bc.read_csv("u_test2", "samples", describe=False) self.assertEqual(len(user.records), 500) def test_read_csv_antenna_id_no_places(self): user = bc.read_csv("u_test_antennas", "samples", describe=False) self.assertEqual(user.records[1], Record(interaction='call', direction='in', correspondent_id='770000001', datetime=datetime.datetime(2013, 12, 16, 5, 39, 30), call_duration=0, position=Position('13084', None))) result = {'allweek': {'allday': None}} self.assertEqual(bc.spatial.radius_of_gyration(user, groupby=None), result) def test_read_csv_antenna_id(self): user = bc.read_csv("u_test_antennas", "samples", antennas_path="samples/towers.csv", describe=False) self.assertEqual(user.records[1], Record(interaction='call', direction='in', correspondent_id='770000001', datetime=datetime.datetime(2013, 12, 16, 5, 39, 30), call_duration=0, position=Position('13084', None))) radius = bc.spatial.radius_of_gyration(user, groupby=None)['allweek']['allday'] self.assertGreater(radius, 0) def test_read_csv_no_position(self): user = bc.read_csv("u_test_no_position", "samples", describe=False) self.assertEqual(user.records[1], Record(interaction='call', direction='in', correspondent_id='770000001', datetime=datetime.datetime(2013, 12, 16, 5, 39, 30), call_duration=0, position=Position())) def test_read_csv_attributes(self): user = bc.read_csv("u_test2", "samples", attributes_path="samples/attributes", describe=False) self.assertEqual(user.attributes, { 'gender': 'male', 'age': '42', 'is_subscriber': 'True', 'individual_id': '7atr8f53fg41' }) bandicoot-0.4.0/bandicoot/tests/test_regressions.py0000644000076500000240000000666712545050666023165 0ustar cynddlstaff00000000000000""" Automatic regression tests """ import bandicoot as bc import unittest from testing_tools import parse_dict, metric_suite import os ARGS = {'flatten': True, 'summary': 'extended', 'split_week': True, 'split_day': True} class TestRegressions(unittest.TestCase): def setUp(self): if not getattr(TestRegressions, '_dir_changed', False): abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestRegressions._dir_changed = True self.empty_user = bc.User() self.empty_user.attributes['empty'] = True self.sample_user = bc.tests.generate_user.sample_user() self.network_ego = bc.read_csv('ego', 'samples/network', 'samples/towers.csv', attributes_path='samples/attributes', network=True, warnings=False, describe=False) # Manual users self.user_a = bc.read_csv('A', 'samples/manual', 'samples/towers.csv', network=False, warnings=False, describe=False) self.user_a_network = bc.read_csv('A', 'samples/manual', 'samples/towers.csv', attributes_path='samples/attributes', network=True, warnings=False, describe=False) self.user_a_orange = bc.io.read_orange('A_orange', 'samples/manual', network=False, warnings=False, describe=False) self.user_a_orange_network = bc.io.read_orange('A_orange', 'samples/manual', network=True, attributes_path='samples/attributes', warnings=False, describe=False) def test_empty_user_all(self): self.assertTrue(*metric_suite(self.empty_user, parse_dict("samples/regressions/empty_user.json")['null'], **ARGS)) def test_sample_user(self): self.assertTrue(*metric_suite(self.sample_user, parse_dict("samples/regressions/sample_user.json")['sample_user'], groupby=None, **ARGS)) def test_network_ego(self): self.assertTrue(*metric_suite(self.network_ego, parse_dict("samples/regressions/ego.json")['ego'], **ARGS)) def test_manual(self): result = parse_dict("samples/regressions/manual_a.json")['A'] result.pop('name') result.pop('reporting__antennas_path') network_result = parse_dict("samples/regressions/manual_a_orange_network.json")['A_orange'] network_result.pop('name') network_result.pop('reporting__antennas_path') self.assertTrue(*metric_suite(self.user_a, result, **ARGS)) self.assertTrue(*metric_suite(self.user_a_orange, result, **ARGS)) self.assertTrue(*metric_suite(self.user_a_orange_network, network_result, network=True, **ARGS)) self.assertTrue(*metric_suite(self.user_a_network, network_result, network=True, **ARGS)) def _generate(self): bc.io.to_json(bc.utils.all(self.empty_user, **ARGS), 'samples/regressions/empty_user.json') bc.io.to_json(bc.utils.all(self.sample_user, groupby=None, **ARGS), 'samples/regressions/sample_user.json') bc.io.to_json(bc.utils.all(self.network_ego, **ARGS), 'samples/regressions/ego.json') bc.io.to_json(bc.utils.all(self.user_a, network=True, **ARGS), 'samples/regressions/manual_a.json') bc.io.to_json(bc.utils.all(self.user_a_orange_network, network=True, **ARGS), 'samples/regressions/manual_a_orange_network.json') if __name__ == '__main__': t = TestRegressions('_generate') t.setUp() t._generate() bandicoot-0.4.0/bandicoot/tests/test_sequences.py0000644000076500000240000000245212545050666022601 0ustar cynddlstaff00000000000000import unittest import bandicoot as bc from bandicoot.tests.generate_user import random_burst from datetime import timedelta class InterEventTests(unittest.TestCase): def setUp(self): self.user = bc.User() def test_empty_interevent(self): self.user.records = [] interevent = bc.individual.interevent_time(self.user) self.assertEqual(interevent['allweek']['allday']['text'].values(), [None] * 2) def test_sequence_interevent(self): # 10 minutes burst _delta = timedelta(hours=2, minutes=13) self.user.records = random_burst(100, interaction='text', delta=_delta) interevent = bc.individual.interevent_time(self.user, groupby=None) self.assertEqual(interevent['allweek']['allday']['text']['mean'], _delta.total_seconds()) self.assertEqual(interevent['allweek']['allday']['text']['std'], 0) _delta = timedelta(hours=10, minutes=10) self.user.records = random_burst(100, interaction='text', delta=_delta) interevent = bc.individual.interevent_time(self.user, groupby=None) self.assertEqual(interevent['allweek']['allday']['text']['mean'], _delta.total_seconds()) bandicoot-0.4.0/bandicoot/tests/test_spatial.py0000644000076500000240000000244312545050666022243 0ustar cynddlstaff00000000000000from __future__ import division import unittest import os import bandicoot as bc import numpy as np class TestChurn(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestChurn._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestChurn._dir_changed = True self.user = bc.io.read_csv("churn_user", "samples", describe=False, warnings=False) def test_churn(self): distribution = bc.spatial.churn_rate(self.user, summary=None) v1 = [1 / 3, 1 / 3, 1 / 3, 0] v2 = v1 v3 = [1 / 4, 3 / 4, 0, 0] v4 = [0, 0, 1 / 2, 1 / 2] cos_1 = 1 - np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)) cos_2 = 1 - np.dot(v2, v3) / (np.linalg.norm(v2) * np.linalg.norm(v3)) cos_3 = 1 - np.dot(v3, v4) / (np.linalg.norm(v3) * np.linalg.norm(v4)) self.assertEqual(distribution, [cos_1, cos_2, cos_3]) churn_rate = bc.spatial.churn_rate(self.user) self.assertEqual(churn_rate['mean'], np.mean([cos_1, cos_2, cos_3])) self.assertEqual(churn_rate['std'], np.std([cos_1, cos_2, cos_3])) bandicoot-0.4.0/bandicoot/tests/test_special.py0000644000076500000240000000312712545050666022226 0ustar cynddlstaff00000000000000""" Test special non-core functionality such as the generation of punchcards. """ import bandicoot as bc import unittest from testing_tools import parse_dict, metric_suite import os class TestSpecial(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestSpecial._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestSpecial._dir_changed = True self.user_X = bc.io.read_csv( "X", "samples/special", "samples/special/antennas.csv", describe=False, warnings=False) self.user_Y = bc.io.read_csv( "Y", "samples/special", describe=False, warnings=False) def assertAlmostEqualLists(self, list_a, list_b, places=5): self.assertTrue(len(list_a) == len(list_b)) for i in range(len(list_a)): self.assertAlmostEqual(list_a[i], list_b[i]) def test_X_punchcard(self): self.X_punchcards = bc.special.punchcard.create_punchcards( self.user_X, split_interval=60) self.assertAlmostEqualLists(self.X_punchcards, bc.special.punchcard.read_csv( "samples/special/punchcard_X_60min_interval.csv")) def test_Y_punchcard(self): self.Y_punchcards = bc.special.punchcard.create_punchcards( self.user_Y, split_interval=5) self.assertAlmostEqualLists(self.Y_punchcards, bc.special.punchcard.read_csv( "samples/special/punchcard_Y_5min_interval.csv")) bandicoot-0.4.0/bandicoot/tests/test_utils.py0000644000076500000240000001021412545050666021741 0ustar cynddlstaff00000000000000""" Test the bandicoot.utils module, mostly using the numpy library. """ import bandicoot as bc import unittest from scipy import stats import numpy as np import os class TestUtils(unittest.TestCase): @classmethod def setUpClass(cls): cls._dir_changed = False def setUp(self): if not TestUtils._dir_changed: abspath = os.path.abspath(__file__) name = abspath.index(os.path.basename(__file__)) abspath = abspath[:name] os.chdir(abspath) TestUtils._dir_changed = True self.user = bc.io.read_orange("u_test", "samples", describe=False) self.list_1 = np.random.randint(-1000, 1000, size=10000) self.list_2 = np.random.randint(1, 1000, size=9001) def test_flatten(self): d = {'alpha': 1, 'beta': {'a': 10, 'b': 42}} self.assertEqual(bc.utils.flatten(d), { 'alpha': 1, 'beta__a': 10, 'beta__b': 42}) self.assertEqual(bc.utils.flatten(d, separator='.'), { 'alpha': 1, 'beta.a': 10, 'beta.b': 42}) def test_skewness(self): self.assertEqual(bc.helper.tools.skewness([]), None) self.assertAlmostEqual(bc.helper.tools.skewness([1, 2, 3, 4, 7]), stats.skew([1, 2, 3, 4, 7])) def test_kurtosis(self): self.assertEqual(bc.helper.tools.kurtosis([]), None) self.assertAlmostEqual(bc.helper.tools.kurtosis([1, 2, 3, 4, 5]), stats.kurtosis([1, 2, 3, 4, 5], fisher=False)) self.assertAlmostEqual(bc.helper.tools.kurtosis([1, 6, 6, 6, 9, 17]), stats.kurtosis([1, 6, 6, 6, 9, 17], fisher=False)) self.assertAlmostEqual(bc.helper.tools.kurtosis(self.list_1), stats.kurtosis(self.list_1, fisher=False)) self.assertAlmostEqual(bc.helper.tools.kurtosis(self.list_2), stats.kurtosis(self.list_2, fisher=False)) def test_mean(self): self.assertEqual(bc.helper.tools.mean([]), None) self.assertAlmostEqual(bc.helper.tools.mean(self.list_1), np.average(self.list_1)) self.assertAlmostEqual(bc.helper.tools.mean(self.list_2), np.average(self.list_2)) def test_std(self): self.assertEqual(bc.helper.tools.std([]), None) self.assertAlmostEqual(bc.helper.tools.std(self.list_1), np.std(self.list_1)) self.assertAlmostEqual(bc.helper.tools.std(self.list_2), np.std(self.list_2)) def test_median(self): self.assertEqual(bc.helper.tools.median([]), None) self.assertEqual(bc.helper.tools.median(self.list_1), np.median(self.list_1)) self.assertEqual(bc.helper.tools.median(self.list_2), np.median(self.list_2)) def test_entropy(self): self.assertEqual(bc.helper.tools.entropy([]), None) self.assertAlmostEqual(bc.helper.tools.entropy(self.list_2), stats.entropy(self.list_2)) def test_great_circle_distance(self): pt1 = [-1, -1] pt2 = [1, 1] pt3 = [0, 0] pt4 = [100, -100] self.assertEqual(bc.helper.tools.great_circle_distance(pt1, pt2), 0) self.assertEqual(bc.helper.tools.great_circle_distance(pt1, pt3), 0) self.assertEqual(bc.helper.tools.great_circle_distance(pt1, pt4), 20015.086796020572) self.assertEqual(bc.helper.tools.great_circle_distance(pt2, pt3), 0) self.assertEqual(bc.helper.tools.great_circle_distance(pt2, pt4), 20015.086796020572) self.assertEqual(bc.helper.tools.great_circle_distance(pt3, pt4), 20015.086796020572) def test_summary_stats(self): self.assertEqual(bc.helper.tools.summary_stats([1, 5, 2]), bc.helper.tools.SummaryStats(mean=2.6666666666666665, std=1.699673171197595, min=1.0, max=5.0, median=2.0, skewness=0.5280049792181879, kurtosis=1.4999999999999998, distribution=[1, 2, 5])) self.assertEqual(bc.helper.tools.summary_stats([1, 2, 3]), bc.helper.tools.SummaryStats(mean=2.0, std=0.816496580927726, min=1.0, max=3.0, median=2.0, skewness=0.0, kurtosis=1.5, distribution=[1, 2, 3])) self.assertEqual(bc.helper.tools.summary_stats([]), bc.helper.tools.SummaryStats(mean=0., std=0., min=0., max=0., median=0., skewness=0., kurtosis=0., distribution=[])) def test_all(self): pass bandicoot-0.4.0/bandicoot/tests/testing_tools.py0000644000076500000240000000371612545050666022450 0ustar cynddlstaff00000000000000import json import filecmp import bandicoot as bc import numpy as np def _convert(input): """ Recursive convertion of unicode """ if isinstance(input, dict): return {_convert(key): _convert(value) for key, value in input.iteritems()} elif isinstance(input, list): return [_convert(element) for element in input] elif isinstance(input, unicode): return input.encode('utf-8') else: return input def parse_dict(path): data = open(path) dict_data = _convert(json.load(data)) return dict_data def file_equality(f1, f2): '''Returns true if the files are the same.''' return filecmp.cmp(f1, f2) def metric_suite(user, answers, decimal=7, **kwargs): ''' Runs the complete metric suite. If any of the metrics is different than the expected answer, return False. ''' results = bc.utils.all(user, **kwargs) test_result, msg = compare_dict(answers, results, decimal=decimal) return test_result, msg def compare_dict(answer, result, decimal=7): '''Returns true if two dictionaries are approximately equal. Returns false otherwise.''' flat_answer = bc.utils.flatten(answer) flat_result = bc.utils.flatten(result) for key in flat_answer.keys(): if key not in flat_result.keys(): return False, "The key {} was not there.".format(key) answer_v, result_v = flat_answer[key], flat_result[key] if isinstance(answer_v, (float, int)) and isinstance(result_v, (float, int)): try: np.testing.assert_almost_equal(answer_v, result_v, decimal=decimal) except AssertionError: err_msg = "The key {} produced a different result: expected {}, got {}.".format(key, answer_v, result_v) return False, err_msg elif answer_v != result_v: return False, "The key {} produced a different result: expected {}, got {}.".format(key, answer_v, result_v) return True, "" bandicoot-0.4.0/bandicoot/utils.py0000755000076500000240000002142012545050666017544 0ustar cynddlstaff00000000000000from bandicoot.helper.tools import OrderedDict, warning_str from bandicoot.helper.group import group_records, DATE_GROUPERS import bandicoot as bc from functools import partial def flatten(d, parent_key='', separator='__'): """ Flatten a nested dictionary. Parameters ---------- d: dict_like Dictionary to flatten. parent_key: string, optional Concatenated names of the parent keys. separator: string, optional Separator between the names of the each key. The default separator is '_'. Examples -------- >>> d = {'alpha': 1, 'beta': {'a': 10, 'b': 42}} >>> flatten(d) == {'alpha': 1, 'beta_a': 10, 'beta_b': 42} True >>> flatten(d, separator='.') == {'alpha': 1, 'beta.a': 10, 'beta.b': 42} True """ items = [] for k, v in d.items(): new_key = parent_key + separator + k if parent_key else k if isinstance(v, dict): items.extend(flatten(v, new_key, separator).items()) else: items.append((new_key, v)) return OrderedDict(items) def all(user, groupby='week', summary='default', network=False, split_week=False, split_day=False, attributes=True, flatten=False): """ Returns a dictionary containing all bandicoot indicators for the user, as well as reporting variables. Relevant indicators are defined in the 'individual', and 'spatial' modules. =================================== ======================================================================= Reporting variables Description =================================== ======================================================================= antennas_path path of the CSV file containing antennas locations attributes_path directory where attributes were loaded version bandicoot version groupby grouping method ('week' or None) split_week whether or not indicators are also computed for weekday and weekend split_day whether or not indicators are also computed for day and night start_time time of the first record end_time time of the last record night_start, night_end start and end time to define nights weekend days used to define the weekend (``[6, 7]`` by default, where 1 is Monday) bins number of weeks if the record are grouped has_call whether or not records include calls has_text whether or not records include texts has_home whether or not a :meth:`home location ` has been found has_network whether or not correspondents where loaded percent_records_missing_location percentage of records without location antennas_missing_locations number of antennas missing a location percent_outofnetwork_calls percentage of calls, received or emitted, made with a correspondant not loaded in the network percent_outofnetwork_texts percentage of texts with contacts not loaded in the network percent_outofnetwork_contacts percentage of contacts not loaded in the network percent_outofnetwork_call_durations percentage of minutes of calls where the contact was not loaded in the network number_of_records total number of records =================================== ======================================================================= We also include a last set of reporting variables, for the records ignored at load-time. Values can be ignored due to missing or inconsistent fields (e.g., not including a valid 'datetime' value). .. code-block:: python { 'all': 0, 'interaction': 0, 'direction': 0, 'correspondent_id': 0, 'datetime': 0, 'call_duration': 0 } with the total number of records ignored (key ``'all'``), as well as the number of records with faulty values for each columns. """ # Warn the user if they are selecting weekly and there's only one week if groupby is not None: if len(set(DATE_GROUPERS[groupby](r.datetime) for r in user.records)) <= 1: print warning_str('Grouping by week, but all data is from the same week!') scalar_type = 'distribution_scalar' if groupby == 'week' else 'scalar' summary_type = 'distribution_summarystats' if groupby == 'week' else 'summarystats' number_of_interactions_in = partial(bc.individual.number_of_interactions, direction='in') number_of_interactions_in.__name__ = 'number_of_interaction_in' number_of_interactions_out = partial(bc.individual.number_of_interactions, direction='out') number_of_interactions_out.__name__ = 'number_of_interaction_out' functions = [ (bc.individual.active_days, scalar_type), (bc.individual.number_of_contacts, scalar_type), (bc.individual.call_duration, summary_type), (bc.individual.percent_nocturnal, scalar_type), (bc.individual.percent_initiated_conversations, scalar_type), (bc.individual.percent_initiated_interactions, scalar_type), (bc.individual.response_delay_text, summary_type), (bc.individual.response_rate_text, scalar_type), (bc.individual.entropy_of_contacts, scalar_type), (bc.individual.balance_of_contacts, summary_type), (bc.individual.interactions_per_contact, summary_type), (bc.individual.interevent_time, summary_type), (bc.individual.percent_pareto_interactions, scalar_type), (bc.individual.percent_pareto_durations, scalar_type), (bc.individual.number_of_interactions, scalar_type), (number_of_interactions_in, scalar_type), (number_of_interactions_out, scalar_type), (bc.spatial.number_of_antennas, scalar_type), (bc.spatial.entropy_of_antennas, scalar_type), (bc.spatial.percent_at_home, scalar_type), (bc.spatial.radius_of_gyration, scalar_type), (bc.spatial.frequent_antennas, scalar_type), (bc.spatial.churn_rate, scalar_type) ] network_functions = [ bc.network.clustering_coefficient_unweighted, bc.network.clustering_coefficient_weighted, bc.network.assortativity_attributes, bc.network.assortativity_indicators ] groups = [[r for r in g] for g in group_records(user, groupby=groupby)] reporting = OrderedDict([ ('antennas_path', user.antennas_path), ('attributes_path', user.attributes_path), ('version', bc.__version__), ('groupby', groupby), ('split_week', split_week), ('split_day', split_day), ('start_time', user.start_time and str(user.start_time)), ('end_time', user.end_time and str(user.end_time)), ('night_start', str(user.night_start)), ('night_end', str(user.night_end)), ('weekend', user.weekend), ('bins', len(groups)), ('has_call', user.has_call), ('has_text', user.has_text), ('has_home', user.has_home), ('has_network', user.has_network), ('percent_records_missing_location', bc.helper.tools.percent_records_missing_location(user)), ('antennas_missing_locations', bc.helper.tools.antennas_missing_locations(user)), ('percent_outofnetwork_calls', user.percent_outofnetwork_calls), ('percent_outofnetwork_texts', user.percent_outofnetwork_texts), ('percent_outofnetwork_contacts', user.percent_outofnetwork_contacts), ('percent_outofnetwork_call_durations', user.percent_outofnetwork_call_durations), ]) if user.records is not None: reporting['number_of_records'] = len(user.records) else: reporting['number_of_records'] = 0. if user.ignored_records is not None: reporting['ignored_records'] = user.ignored_records returned = OrderedDict([ ('name', user.name), ('reporting', reporting) ]) for fun, datatype in functions: try: metric = fun(user, groupby=groupby, summary=summary, datatype=datatype, split_week=split_week, split_day=split_day) except ValueError: metric = fun(user, groupby=groupby, datatype=datatype, split_week=split_week, split_day=split_day) returned[fun.__name__] = metric if network and user.has_network: for fun in network_functions: returned[fun.__name__] = fun(user) if attributes and user.attributes != {}: returned['attributes'] = user.attributes if flatten is True: return globals()['flatten'](returned) return returned bandicoot-0.4.0/bandicoot.egg-info/0000755000076500000240000000000012545051250017510 5ustar cynddlstaff00000000000000bandicoot-0.4.0/bandicoot.egg-info/dependency_links.txt0000644000076500000240000000000112545051247023564 0ustar cynddlstaff00000000000000 bandicoot-0.4.0/bandicoot.egg-info/pbr.json0000644000076500000240000000005712545051247021176 0ustar cynddlstaff00000000000000{"is_release": false, "git_version": "a933fbd"}bandicoot-0.4.0/bandicoot.egg-info/PKG-INFO0000644000076500000240000000126612545051247020620 0ustar cynddlstaff00000000000000Metadata-Version: 1.1 Name: bandicoot Version: 0.4.0 Summary: A toolbox to analyze mobile phone metadata. Home-page: https://github.com/yvesalexandre/bandicoot Author: Yves-Alexandre de Montjoye Author-email: yvesalexandre@demontjoye.com License: MIT Description: UNKNOWN Platform: UNKNOWN Classifier: Environment :: Plugins Classifier: Intended Audience :: Science/Research Classifier: License :: OSI Approved :: MIT License Classifier: Programming Language :: Python Classifier: Programming Language :: Python :: 2 Classifier: Programming Language :: Python :: 2.7 Classifier: Topic :: Scientific/Engineering :: Information Analysis Classifier: Topic :: Scientific/Engineering :: Mathematics bandicoot-0.4.0/bandicoot.egg-info/SOURCES.txt0000644000076500000240000000163012545051247021402 0ustar cynddlstaff00000000000000setup.py bandicoot/__init__.py bandicoot/core.py bandicoot/individual.py bandicoot/io.py bandicoot/network.py bandicoot/spatial.py bandicoot/utils.py bandicoot.egg-info/PKG-INFO bandicoot.egg-info/SOURCES.txt bandicoot.egg-info/dependency_links.txt bandicoot.egg-info/pbr.json bandicoot.egg-info/top_level.txt bandicoot/helper/__init__.py bandicoot/helper/group.py bandicoot/helper/tools.py bandicoot/special/__init__.py bandicoot/special/demo.py bandicoot/special/punchcard.py bandicoot/tests/__init__.py bandicoot/tests/generate_user.py bandicoot/tests/test_core.py bandicoot/tests/test_export.py bandicoot/tests/test_group.py bandicoot/tests/test_manual.py bandicoot/tests/test_network.py bandicoot/tests/test_parsers.py bandicoot/tests/test_regressions.py bandicoot/tests/test_sequences.py bandicoot/tests/test_spatial.py bandicoot/tests/test_special.py bandicoot/tests/test_utils.py bandicoot/tests/testing_tools.pybandicoot-0.4.0/bandicoot.egg-info/top_level.txt0000644000076500000240000000001212545051247022241 0ustar cynddlstaff00000000000000bandicoot bandicoot-0.4.0/PKG-INFO0000644000076500000240000000126612545051250015156 0ustar cynddlstaff00000000000000Metadata-Version: 1.1 Name: bandicoot Version: 0.4.0 Summary: A toolbox to analyze mobile phone metadata. Home-page: https://github.com/yvesalexandre/bandicoot Author: Yves-Alexandre de Montjoye Author-email: yvesalexandre@demontjoye.com License: MIT Description: UNKNOWN Platform: UNKNOWN Classifier: Environment :: Plugins Classifier: Intended Audience :: Science/Research Classifier: License :: OSI Approved :: MIT License Classifier: Programming Language :: Python Classifier: Programming Language :: Python :: 2 Classifier: Programming Language :: Python :: 2.7 Classifier: Topic :: Scientific/Engineering :: Information Analysis Classifier: Topic :: Scientific/Engineering :: Mathematics bandicoot-0.4.0/setup.cfg0000644000076500000240000000007312545051250015675 0ustar cynddlstaff00000000000000[egg_info] tag_build = tag_date = 0 tag_svn_revision = 0 bandicoot-0.4.0/setup.py0000644000076500000240000000211212545050666015574 0ustar cynddlstaff00000000000000#!/usr/bin/env python from setuptools import setup import re import ast _version_re = re.compile(r'__version__\s+=\s+(.*)') with open('bandicoot/__init__.py', 'rb') as f: version = str(ast.literal_eval(_version_re.search( f.read().decode('utf-8')).group(1))) setup( name='bandicoot', author='Yves-Alexandre de Montjoye', author_email='yvesalexandre@demontjoye.com', version=version, url="https://github.com/yvesalexandre/bandicoot", license="MIT", packages=[ 'bandicoot', 'bandicoot.helper', 'bandicoot.tests', 'bandicoot.special' ], description="A toolbox to analyze mobile phone metadata.", classifiers=[ 'Environment :: Plugins', 'Intended Audience :: Science/Research', 'License :: OSI Approved :: MIT License', 'Programming Language :: Python', 'Programming Language :: Python :: 2', 'Programming Language :: Python :: 2.7', 'Topic :: Scientific/Engineering :: Information Analysis', 'Topic :: Scientific/Engineering :: Mathematics' ])