# By: Riasat Ullah # This file contains methods for handling live call related analytics. from analytics import analytics_tools from dbqueries import db_billings, db_live_call_routing from utils import constants, errors, helpers, var_names from utils.communication_vendors import Twilio import configuration as configs import numpy import pandas class LiveCallAnalytics(object): def __init__(self, conn, timestamp, organization_id, start_date, end_date, call_routing_refs=None, service_refs=None, team_refs=None, call_logs=None, with_mappings=True, with_volume_discount=False): if end_date < start_date: raise ValueError(errors.err_time_end_before_start) self.conn = conn self.timestamp = timestamp self.organization_id = organization_id self.start_date = start_date self.end_date = end_date self.call_routing_refs = call_routing_refs self.service_refs = service_refs self.team_refs = team_refs self.call_logs = call_logs self.with_volume_discount = with_volume_discount # This will be used (seconds) if there is no data available to calculate the call duration self.default_call_duration_seconds = 60 if self.call_logs is None: org_logs_dict = db_live_call_routing.get_live_call_routing_analytics_details( self.conn, self.start_date, self.end_date, self.organization_id, call_routing_refs=self.call_routing_refs, service_refs=self.service_refs, team_refs=self.team_refs ) self.call_logs = org_logs_dict[self.organization_id] if self.organization_id in org_logs_dict else [] self.org_routings, self.org_services, self.org_users =\ db_live_call_routing.get_current_component_names_for_call_routing_analytics( self.conn, self.timestamp, self.start_date, self.organization_id)\ if with_mappings else (None, None, None) self.country_rates = db_live_call_routing.get_live_call_routing_country_specific_rates( self.conn, self.start_date) self.logs_df = pandas.DataFrame( self.call_logs, columns=[var_names.log_id, var_names.vendor_name, var_names.vendor_id, var_names.call_time, var_names.call_start_timestamp, var_names.call_end_timestamp, var_names.call_routing_id, var_names.from_iso, var_names.from_number, var_names.to_iso, var_names.to_number, var_names.call_status, var_names.recording_url, var_names.answer_timestamp, var_names.answered_by, var_names.forwarding_count, var_names.service_id, var_names.organization_instance_id, var_names.events] ) self.forwarding_df = pandas.DataFrame( self.standardize_call_forwarding_data(), columns=[var_names.log_id, var_names.organization_instance_id, var_names.user_id, var_names.iso_country_code, var_names.call_leg_time, var_names.acknowledgement_time, var_names.answer_duration, var_names.call_leg_price] ) def get_max_minute_iso_and_rate(self, iso_list): ''' Get the maximum minute rate to apply on a call leg given the inbound and outbound calling countries. :param iso_list: (list) of iso code of countries :return: (tuple) -> iso code, maximum rate (float) ''' max_iso, max_rate = 'US', 0.1 for iso in iso_list: if iso in self.country_rates: iso_rate = self.country_rates[iso][var_names.minute_rate] if iso_rate > max_rate: max_iso = iso max_rate = iso_rate return max_iso, max_rate def standardize_call_forwarding_data(self): ''' This function parses through the call logs, filters out call forwarding related instance events and puts them in an iterable list format. :return: (list of dict) of call forwarding information ''' data = [] for item in self.call_logs: inst_id = item[var_names.organization_instance_id] inst_events = item[var_names.events] if item[var_names.call_status] == constants.suppressed_state: continue else: if inst_events is None: leg_time = None leg_iso, leg_rate = self.get_max_minute_iso_and_rate( [item[var_names.from_iso], item[var_names.to_iso]]) if item[var_names.call_end_timestamp] is None or item[var_names.call_start_timestamp] is None: if item[var_names.vendor_name] == constants.twilio: leg_time = Twilio(Twilio.get_authentication()).get_call_duration( item[var_names.vendor_id]) / 60 else: leg_time = (item[var_names.call_end_timestamp] - item[var_names.call_start_timestamp]).seconds / 60 if leg_time is None: leg_time = self.default_call_duration_seconds / 60 data.append({var_names.log_id: item[var_names.log_id], var_names.organization_instance_id: inst_id, var_names.user_id: None, var_names.iso_country_code: leg_iso, var_names.acknowledgement_time: None, var_names.answer_duration: None, var_names.call_leg_time: leg_time, var_names.call_leg_price: leg_rate}) else: for i in range(0, len(inst_events) - 1): this_event = inst_events[i] next_event = inst_events[i + 1] this_type, this_tmsp = this_event[var_names.event_type], this_event[var_names.event_timestamp] next_type, next_tmsp = next_event[var_names.event_type], next_event[var_names.event_timestamp] fwd, ack_time, ans_time, leg_time, leg_iso, leg_rate = None, None, None, None, None, 0.1 if i == 0: fwd, ack_time, ans_time = None, None, None leg_iso, leg_rate = self.get_max_minute_iso_and_rate( [item[var_names.from_iso], item[var_names.to_iso]]) leg_time = (this_tmsp - item[var_names.call_start_timestamp]).seconds / 60 data.append({var_names.log_id: item[var_names.log_id], var_names.organization_instance_id: inst_id, var_names.user_id: fwd, var_names.iso_country_code: leg_iso, var_names.acknowledgement_time: ack_time, var_names.answer_duration: ans_time, var_names.call_leg_time: leg_time, var_names.call_leg_price: leg_rate}) if this_type == constants.call_forwarding_event: leg_iso, leg_rate = self.get_max_minute_iso_and_rate( [item[var_names.from_iso], this_event[var_names.iso_country_code]]) fwd = this_event[var_names.forward_to] fwd_time = (next_tmsp - this_tmsp).seconds/60 leg_time = fwd_time if next_type == constants.call_answered_event: ack_time = fwd_time ans_time = (item[var_names.call_end_timestamp] - next_tmsp).seconds/60 leg_time = fwd_time + ans_time elif next_type == constants.call_voicemail_prompt_event: vml_time = (item[var_names.call_end_timestamp] - next_tmsp).seconds/60 leg_time = fwd_time + vml_time else: fwd, ack_time, ans_time = None, None, None leg_iso, leg_rate = self.get_max_minute_iso_and_rate( [item[var_names.from_iso], item[var_names.to_iso]]) leg_time = (next_tmsp - this_tmsp).seconds / 60 if leg_iso is not None: data.append({var_names.log_id: item[var_names.log_id], var_names.organization_instance_id: inst_id, var_names.user_id: fwd, var_names.iso_country_code: leg_iso, var_names.acknowledgement_time: ack_time, var_names.answer_duration: ans_time, var_names.call_leg_time: leg_time, var_names.call_leg_price: leg_rate}) return data def get_user_call_count(self, user_id): ''' Get the number of calls that were made to a user. :param user_id: (int) user ID :return: (int) call count ''' return int(analytics_tools.convert_nan_to_number( self.forwarding_df[self.forwarding_df[var_names.user_id] == user_id][var_names.log_id].count(), rounding=None )) def get_user_answer_count(self, user_id): ''' Get the number of calls that were answered by a user. :param user_id: (int) user ID :return: (int) answered call count ''' return int(analytics_tools.convert_nan_to_number( self.forwarding_df[(self.forwarding_df[var_names.user_id] == user_id) & (self.forwarding_df[var_names.acknowledgement_time].notnull())][ var_names.log_id].count(), rounding=None )) def get_user_avg_acknowledgement_time(self, user_id): ''' Get the average number of minutes taken by a user to receive a call. :param user_id: (int) user ID :return: (float) minutes ''' return analytics_tools.convert_nan_to_number( self.forwarding_df[(self.forwarding_df[var_names.user_id] == user_id) & (self.forwarding_df[var_names.acknowledgement_time].notnull())][ var_names.acknowledgement_time].mean() ) def get_user_avg_answer_duration(self, user_id): ''' Get the average number of minutes a user's answer lasts. :param user_id: (int) user ID :return: (float) minutes ''' return analytics_tools.convert_nan_to_number( self.forwarding_df[(self.forwarding_df[var_names.user_id] == user_id) & (self.forwarding_df[var_names.answer_duration].notnull())][ var_names.answer_duration].mean() ) def get_metrics(self): ''' Get the final live call metric details in the format that the web platform expects it. :return: (dict) of metric data ''' data = { var_names.usage: self.org_usage_summary(), var_names.call_logs: self.org_call_logs(), var_names.effectiveness_metrics: self.user_call_metrics(), var_names.count: { var_names.daily: self.get_daily_call_count_graph_data(), var_names.hourly: self.get_hourly_call_count_graph_data() } } return data def get_daily_call_count_graph_data(self): ''' Get the labels and values needed to draw a daily call count graph. :return: (dict) {labels: [...], data: [...]} ''' if len(self.logs_df) > 0: str_date = 'date' self.logs_df[str_date] = self.logs_df[var_names.call_time].dt.date daily_data = self.logs_df.groupby(str_date)[var_names.log_id].count().to_dict() del self.logs_df[str_date] else: daily_data = dict() labels, values = analytics_tools.get_period_labels_and_values(self.start_date, self.end_date, daily_data) return {var_names.labels: labels, var_names.data: values} def get_hourly_call_count_graph_data(self): ''' Get the labels and values needed to draw a hourly call count graph. :return: (dict) {labels: [...], data: [...]} ''' if len(self.logs_df) > 0: str_hour = 'hour' self.logs_df[str_hour] = self.logs_df[var_names.call_time].dt.hour hourly_data = self.logs_df.groupby(str_hour)[var_names.log_id].count().to_dict() del self.logs_df[str_hour] else: hourly_data = dict() labels, values = [], [] for i in range(0, 24): labels.append(i) values.append(hourly_data[i] if i in hourly_data else 0) return {var_names.labels: labels, var_names.data: values} def org_usage_summary(self): ''' Get the summary of live calls at the organization level. :return: (dict) -> summary of usage ''' call_count = len(self.logs_df) answered_count = len(self.logs_df[self.logs_df[var_names.call_status] == constants.answered_state]) answered_perc = round(answered_count/call_count * 100) if call_count > 0 else 0 suppressed_count = len(self.logs_df[self.logs_df[var_names.call_status] == constants.suppressed_state]) missed_count = call_count - answered_count - suppressed_count missed_perc = (100 - answered_perc) if call_count > 0 else 0 avg_answer_minutes = analytics_tools.convert_nan_to_number(( (self.logs_df[var_names.call_end_timestamp].astype('datetime64[ns]') - self.logs_df[var_names.answer_timestamp].astype('datetime64[ns]')) / numpy.timedelta64(1, 's')/60 ).where(self.logs_df[var_names.call_status] == constants.answered_state).mean()) call_recording_minutes = int(analytics_tools.convert_nan_to_number(( numpy.ceil( (self.logs_df[var_names.call_end_timestamp].astype('datetime64[ns]') - self.logs_df[var_names.answer_timestamp].astype('datetime64[ns]')) / numpy.timedelta64(1, 's')/60 )).where((self.logs_df[var_names.call_status] == constants.answered_state) & (~self.logs_df[var_names.recording_url].isnull())).mean())) # US/CA minutes us_ca_minutes = int(analytics_tools.convert_nan_to_number(numpy.ceil( self.forwarding_df.where(self.forwarding_df[var_names.iso_country_code].isin(["US", "CA"])) .groupby(var_names.log_id)[var_names.call_leg_time].sum() ).sum())) us_ca_cost = round(us_ca_minutes * configs.call_routing_minutes_price_us_ca, 2) # international minutes intl_minutes = int(analytics_tools.convert_nan_to_number(numpy.ceil( self.forwarding_df.where(~self.forwarding_df[var_names.iso_country_code].isin(["US", "CA"])) .groupby(var_names.log_id)[var_names.call_leg_time].sum() ).sum())) intl_cost = round(intl_minutes * configs.call_routing_minutes_price_international, 2) # total minutes total_minutes = us_ca_minutes + intl_minutes total_cost = round(us_ca_cost + intl_cost, 2) disc_id, disc_desc, disc = None, None, 0 if self.with_volume_discount and us_ca_minutes >= configs.call_routing_us_min_minutes_for_volume_discount: lcr_vol_disc = db_billings.get_discounts( self.conn, self.timestamp, discount_type=[constants.live_call_routing_volume_discount_type], map_on_org=False ) if len(lcr_vol_disc) > 0: lcr_vol_disc = lcr_vol_disc[0] all_tiers = lcr_vol_disc[var_names.additional_info] for tier in all_tiers: if float(tier[var_names.usage][0]) <= us_ca_minutes < float(tier[var_names.usage][1]): tier_perc = tier[var_names.discount_percent] disc_id = lcr_vol_disc[var_names.discount_id] disc_desc = lcr_vol_disc[var_names.reason] + ' - (' + str(tier_perc) + '%)' disc = round((us_ca_cost * float(tier_perc))/100, 2) data = { var_names.call_count: call_count, var_names.answered_call_count: [answered_count, answered_perc], var_names.missed_call_count: [missed_count, missed_perc], var_names.answer_duration: avg_answer_minutes, var_names.domestic_minutes: us_ca_minutes, var_names.domestic_cost: us_ca_cost, var_names.international_minutes: intl_minutes, var_names.international_cost: intl_cost, var_names.call_duration: total_minutes, var_names.total_cost: total_cost, var_names.recording_minutes: call_recording_minutes, var_names.recording_cost: call_recording_minutes * configs.call_recording_fee_per_minute, var_names.volume_discount: [disc_id, disc_desc, disc] if disc > 0 else None } return data def org_call_logs(self): ''' Get the call logs of the organization. :return: (list of dict) of call logs ''' data = [] for item in self.call_logs: call_duration = (item[var_names.call_end_timestamp] - item[var_names.call_start_timestamp]).seconds/60 time_to_answer, ans_duration = None, None if item[var_names.call_status] == constants.answered_state: time_to_answer = (item[var_names.answer_timestamp] - item[var_names.call_start_timestamp]).seconds/60 ans_duration = (item[var_names.call_end_timestamp] - item[var_names.answer_timestamp]).seconds/60 rout_id, serv_id, user_id =\ item[var_names.call_routing_id], item[var_names.service_id], item[var_names.answered_by] if rout_id is not None and rout_id not in self.org_routings: raise KeyError('Routing ID is not in organization - ' + str(rout_id)) if serv_id is not None and serv_id not in self.org_services: raise KeyError('Service ID is not in organization - ' + str(serv_id)) if user_id is not None and user_id not in self.org_users: raise KeyError('User ID not in in organization - ' + str(user_id)) data.append({ var_names.call_time: item[var_names.call_time], var_names.from_number: item[var_names.from_number], var_names.to_number: item[var_names.to_number], var_names.call_routing: [self.org_routings[rout_id][var_names.routing_name], self.org_routings[rout_id][var_names.call_routing_ref_id]] if rout_id is not None else None, var_names.service: [self.org_services[serv_id][var_names.service_name], self.org_services[serv_id][var_names.service_ref_id]] if serv_id is not None else None, var_names.call_status: item[var_names.call_status], var_names.call_duration: call_duration, var_names.acknowledgement_time: time_to_answer, var_names.answer_duration: ans_duration, var_names.answered_by: [self.org_users[user_id][var_names.name], self.org_users[user_id][var_names.preferred_username]] if user_id is not None else None, var_names.forwarding_count: item[var_names.forwarding_count], var_names.organization_instance_id: item[var_names.organization_instance_id] }) data = helpers.sorted_list_of_dict(data, var_names.call_time, descending=True) return data def user_call_metrics(self): ''' Get the call response and performance of the users. :return: (list of dict) of user metrics ''' data = [] for id_ in sorted(self.org_users.keys()): usr_info = self.org_users[id_] total_calls = self.get_user_call_count(id_) answered_calls = self.get_user_answer_count(id_) missed_calls = total_calls - answered_calls data.append({ var_names.name: usr_info[var_names.name], var_names.preferred_username: usr_info[var_names.preferred_username], var_names.call_count: total_calls, var_names.answered_call_count: answered_calls, var_names.missed_call_count: missed_calls, var_names.acknowledgement_time: self.get_user_avg_acknowledgement_time(id_), var_names.answer_duration: self.get_user_avg_answer_duration(id_), var_names.responsiveness: round(answered_calls/total_calls * 100) if total_calls > 0 else 0 }) data = helpers.sorted_list_of_dict(data, var_names.responsiveness, descending=True) return data