# By: Riasat Ullah # This file contains methods for handling user related analysis and metrics. from analytics import analytics_tools from analytics.instance_analyzer import InstanceAnalyzer from dbqueries import db_analytics, db_members from utils import constants, helpers, var_names import configuration as configs class UserAnalytics(object): def __init__(self, conn, timestamp, organization_id, start_date, end_date, user_id=None, tags=None): self.conn = conn self.organization_id = organization_id self.start_date = start_date self.end_date = end_date self.user_id = user_id self.tags = tags self.members = db_members.get_non_stakeholder_internal_details( conn, timestamp, self.organization_id, start_period=self.start_date, end_period=self.end_date, user_id=self.user_id ) self.user_pol_id = None if user_id is not None: self.user_pol_id = self.members[0][3] self.instances = db_analytics.get_instances_for_analysis( conn, self.start_date, self.end_date, self.organization_id, user_pol_id=self.user_pol_id, tags=self.tags) @staticmethod def get_user_performance_metrics(analyzer): ''' Get the incident acknowledgement and resolution metrics of a given user. :param analyzer: InstanceAnalyzer object :return: (dict) of metrics ''' inc_count = analyzer.get_aggregate_instance_count() acked_inc_count = analyzer.get_aggregate_count_of_instances_acknowledged_by_user() res_inc_count = analyzer.get_aggregate_count_of_instances_resolved_by_user() return { var_names.incident_count: inc_count, var_names.acknowledged_incident_count: acked_inc_count, var_names.acknowledgement_time: analytics_tools.convert_nan_to_number( analyzer.get_average_user_acknowledgement_time()), var_names.resolved_incident_count: res_inc_count, var_names.resolution_time: analytics_tools.convert_nan_to_number( analyzer.get_average_user_resolution_time()), var_names.responsiveness: round(acked_inc_count/inc_count * 100, 2) if inc_count > 0 else 0, var_names.effectiveness: round(res_inc_count/inc_count * 100, 2) if inc_count > 0 else 0 } @staticmethod def get_user_over_worked_metrics(analyzer): ''' Get the metrics for how overworked a user is being. The figures for how fatigued users are becoming, how often they are being interrupted and how much rest they are getting in between incidents are all provided. :param analyzer: InstanceAnalyzer object :return: (dict) of metrics ''' # get the regular and fatigued incident counts daytime_stats = analyzer.weekly_daytime_instance_stats() after_hours_stats = analyzer.weekly_after_hours_instance_stats() sleep_hours_stats = analyzer.weekly_sleep_hours_instance_stats() # get the average interruption and rest minutes interruption_hours = analytics_tools.convert_nan_to_number( analyzer.get_average_user_interruption_minutes_per_day()) rest_hours = analytics_tools.convert_nan_to_number(analyzer.get_average_rest_minutes_between_instances()) # qualify the interruption and rest hours interruption_status = constants.ideal_metrics if interruption_hours > configs.ideal_daily_interruption_minutes: interruption_status = constants.below_ideal_metrics rest_status = constants.ideal_metrics if rest_hours < configs.ideal_daily_rest_minutes: rest_status = constants.below_ideal_metrics return { var_names.regular_incident_count: [daytime_stats[1], after_hours_stats[1], sleep_hours_stats[1]], var_names.fatigued_incident_count: [daytime_stats[2], after_hours_stats[2], sleep_hours_stats[2]], var_names.interruption_hours: [interruption_hours, interruption_status], var_names.rest_hours: [rest_hours, rest_status] } @staticmethod def get_user_hourly_effectiveness_metrics(analyzer, avg_effectiveness): ''' Get the metrics of how a user's effectiveness to resolve incidents by themselves changes by the hour. :param analyzer: InstanceAnalyzer object :param avg_effectiveness: overall average effectiveness of the user :return: (dict) of metrics ''' hourly_effectiveness = analyzer.get_user_hourly_effectiveness() if avg_effectiveness >= configs.ideal_effectiveness: eff_qualitative = constants.ideal_metrics elif configs.below_ideal_effectiveness <= avg_effectiveness < configs.ideal_effectiveness: eff_qualitative = constants.below_ideal_metrics else: eff_qualitative = constants.far_below_ideal_metrics return { var_names.data: hourly_effectiveness, var_names.effectiveness: [avg_effectiveness, eff_qualitative] } def get_user_workload_in_escalation_policies(self, user_id, user_pol_id): ''' Calculate how much work load a given user is taking in each of the escalation policies that he is part of. Only the escalation policies that had incidents on them are included in teh data. :param user_id: (int) ID of the user :param user_pol_id: (int) policy ID of the user :return: (list of list) -> [[policy name, percentage load], ...] ''' user_esc_pols = db_analytics.get_user_escalation_policy_ids_in_period( self.conn, self.start_date, self.end_date, self.organization_id, user_id ) data = [] if len(user_esc_pols) == 0: return esc_pol_instances = db_analytics.get_instances_for_analysis( self.conn, self.start_date, self.end_date, self.organization_id, esc_pol_ids=[x[1] for x in user_esc_pols] ) analyzer = InstanceAnalyzer(configs.standard_timezone, esc_pol_instances) for pol_name, pol_id in user_esc_pols: analyzer.filter_by_escalation_policy_id(pol_id) esc_inc_count = analyzer.get_aggregate_instance_count() if esc_inc_count > 0: analyzer.filter_by_user_policy_id(user_pol_id) user_inc_count = analyzer.get_aggregate_instance_count() perc_load = round(user_inc_count / esc_inc_count * 100, 2) data.append([pol_name, perc_load]) return data def get_org_users_performance_metrics_summary(self): ''' Get the summary of the basic metrics of all the users of the organization. :return: (list of dict) of metric details ''' data = [] for item in self.members: user_id, pref_name, disp_name, user_pol, user_tz = item analyzer = InstanceAnalyzer(user_tz, self.instances) analyzer.filter_by_user_policy_id(user_pol) analyzer.create_user_acknowledgement_and_resolution_columns(user_id) data.append({**{var_names.display_name: disp_name, var_names.preferred_username: pref_name, var_names.timezone: user_tz}, **self.get_user_performance_metrics(analyzer)}) data = helpers.sorted_list_of_dict(data, var_names.incident_count, descending=True) return data def get_user_specific_detailed_metrics(self): ''' Get detailed analytics of a specific user. :return: (dict of dict) of metric details ''' user_id, pref_name, disp_name, user_pol, user_tz = self.members[0] # set up the analyzer analyzer = InstanceAnalyzer(user_tz, self.instances) analyzer.filter_by_user_policy_id(user_pol) analyzer.create_user_acknowledgement_and_resolution_columns(user_id) analyzer.create_user_interruption_minutes_column(user_pol) analyzer.create_user_rest_minutes_column() analyzer.create_regional_time_column() performance_metrics = self.get_user_performance_metrics(analyzer) data = {**{var_names.display_name: disp_name, var_names.preferred_username: pref_name, var_names.timezone: user_tz}, **performance_metrics, **{var_names.fatigue_metrics: self.get_user_over_worked_metrics(analyzer)}, **{var_names.effectiveness_metrics: self.get_user_hourly_effectiveness_metrics( analyzer, performance_metrics[var_names.effectiveness])}, **{var_names.escalation_policy_workload: self.get_user_workload_in_escalation_policies( user_id, user_pol)} } return data