# By: Riasat Ullah # This file represents and object that handles instance comparisons to find how similar they are. from modules.text_processor import TextProcessor from utils import errors, var_names import numpy class InstanceComparator(object): default_relevance = 0.54 def __init__(self, instances: list, base_instance_id: int): self.instances = instances self.base_instance_id = base_instance_id self.base_index = self.get_base_index() self.base_instance = self.instances[self.base_index] def get_base_index(self): for i in range(0, len(self.instances)): inst = self.instances[i] if inst.instance_id == self.base_instance_id: return i raise LookupError(errors.err_unknown_resource) def get_title_similarity_scores(self): titles = [inst.task.details[var_names.task_title] for inst in self.instances] text_processor = TextProcessor(titles) text_processor.default_pre_process() base_instance_words = text_processor.corpus[self.base_index].split(' ') scores = text_processor.get_similarity_scores(base_instance_words, self.base_index) return scores def get_tag_similarity_scores(self): base_inst_tags = self.base_instance.task.get_tags() if base_inst_tags is None or len(base_inst_tags) == 0: return [1 for i in range(0, len(self.instances))] else: scores = [] for inst in self.instances: comp_inst_tags = inst.task.get_tags() scores.append(len(set(base_inst_tags) & set(comp_inst_tags)) / len(base_inst_tags)) return scores def get_time_similarity_scores(self): base_timestamp = self.base_instance.instance_timestamp base_mins = (base_timestamp.hour * 60) + base_timestamp.minute max_diff = max([abs(0 - base_mins), abs(1440 - base_mins)]) scores = [] for inst in self.instances: inst_mins = (inst.instance_timestamp.hour * 60) + inst.instance_timestamp.minute inst_score = 1 - (abs(inst_mins - base_mins)/max_diff) scores.append(inst_score) return scores def get_date_similarity_scores(self): base_date = self.base_instance.instance_timestamp.date() relevance_period = 180 scores = [] for inst in self.instances: comp_date = inst.instance_timestamp.date() days_diff = abs((base_date - comp_date).days) if days_diff < relevance_period: scores.append((relevance_period - days_diff) / relevance_period) else: scores.append(0) return scores def get_assignee_similarity_scores(self): base_serv_id = self.base_instance.task.details[var_names.service_id] base_integ_id = self.base_instance.task.details[var_names.integration_id] if self.base_instance.task.assignees is None: base_init_policies = [] else: base_init_policies = [item.for_policy_id for item in self.base_instance.task.assignees] scores = [] for inst in self.instances: this_score = 0 if inst.task.details[var_names.service_id] == base_serv_id: this_score += 1 if inst.task.details[var_names.integration_id] == base_integ_id: this_score += 1 # Look at the additional policies (if any) if inst.task.assignees is None: comp_init_policies = [] else: comp_init_policies = [item.for_policy_id for item in inst.task.assignees] # If both the base instance and the comparing instance do not have additional policies # then base the scoring only on the service and integration similarities. if len(base_init_policies) == 0 and len(comp_init_policies) == 0: scores.append(this_score/2) elif (len(base_init_policies) == 0 and len(comp_init_policies) > 0) or\ (len(base_init_policies) > 0 and len(comp_init_policies) == 0): scores.append(this_score/3) else: this_score += len(set(base_init_policies) & set(comp_init_policies)) / len(base_init_policies) scores.append(this_score/3) return scores def get_weighted_combined_scores(self): wg_title_scores = list(map(lambda x: x * 0.60, self.get_title_similarity_scores())) wg_tag_scores = list(map(lambda x: x * 0.15, self.get_tag_similarity_scores())) wg_time_scores = list(map(lambda x: x * 0.10, self.get_time_similarity_scores())) wg_date_scores = list(map(lambda x: x * 0.05, self.get_date_similarity_scores())) wg_assignee_scores = list(map(lambda x: x * 0.10, self.get_assignee_similarity_scores())) return list(numpy.sum( numpy.array([wg_title_scores, wg_tag_scores, wg_time_scores, wg_date_scores, wg_assignee_scores]), 0 )) def get_similar_instances(self, min_relevance=default_relevance): scores = self.get_weighted_combined_scores() similar_incidents = [] for i in range(0, len(scores)): inst_score = scores[i] if inst_score >= min_relevance and i != self.base_index: similar_incidents.append(self.instances[i]) return similar_incidents def get_similar_instance_details(self, min_relevance=default_relevance): scores = self.get_weighted_combined_scores() similar_incidents = [] for i in range(0, len(scores)): inst_score = scores[i] if inst_score >= min_relevance and i != self.base_index: inst = self.instances[i] similar_incidents.append({ var_names.instance_id: inst.instance_id, var_names.organization_instance_id: inst.organization_instance_id, var_names.instance_timestamp: inst.instance_timestamp, var_names.resolution_time: inst.get_resolution_time(), var_names.task_title: inst.task.details[var_names.task_title], var_names.resolved_by: inst.get_resolver(), var_names.urgency_level: inst.task.details[var_names.urgency_level], var_names.similarity_score: inst_score }) return similar_incidents def get_most_similar_instance(self, min_relevance=default_relevance): scores = self.get_weighted_combined_scores() max_score = 0 similar_instance = None for i in range(0, len(scores)): inst_score = scores[i] if inst_score >= min_relevance and inst_score > max_score and i != self.base_index: max_score = inst_score similar_instance = self.instances[i] return similar_instance