nirdizati_light.labeling.common

 1from enum import Enum
 2from nirdizati_light.utils.log_metrics import events_by_date, resources_by_date, new_trace_start
 3from nirdizati_light.utils.time_metrics import elapsed_time_id, remaining_time_id, count_on_event_day, duration
 4from pm4py.objects.log.obj import EventLog, Trace
 5
 6def get_intercase_attributes(log, encoding):
 7    """Dict of kwargs
 8    These intercase attributes are expensive operations!!!
 9    """
10    # Expensive operations
11    executed_events = events_by_date(log) if encoding.add_executed_events else None
12    resources_used = resources_by_date(log) if encoding.add_resources_used else None
13    new_traces = new_trace_start(log) if encoding.add_new_traces else None
14    kwargs = {'executed_events': executed_events, 'resources_used': resources_used, 'new_traces': new_traces}
15    # 'label': label}  TODO: is it really necessary to add this field in the dict?
16    return kwargs
17
18class LabelTypes(Enum):
19    """
20    Possible labels for the predictive task
21    """
22    NEXT_ACTIVITY = 'next_activity'
23    ATTRIBUTE_STRING = 'label_attribute_string'
24    REMAINING_TIME = 'remaining_time'
25    DURATION = 'duration'
26    NO_LABEL = 'no_label'
27
28class ThresholdTypes(Enum):
29    """
30    ThresholdTypes
31    """
32    THRESHOLD_MEAN = 'threshold_mean'
33    THRESHOLD_CUSTOM = 'threshold_custom'
34    NONE = 'none'
35
36
37def add_label_column(trace, labeling_type, prefix_length: int):
38    """TODO COMMENT ME
39    """
40    if labeling_type == LabelTypes.NEXT_ACTIVITY.value:
41        return next_event_name(trace, prefix_length)
42    elif labeling_type == LabelTypes.ATTRIBUTE_STRING.value:
43        try:
44            return trace.attributes['label']
45        except KeyError:
46            return trace[0]['label']
47    elif labeling_type == LabelTypes.REMAINING_TIME.value:
48        return remaining_time_id(trace, prefix_length)
49    elif labeling_type == LabelTypes.DURATION.value:
50        return duration(trace)
51    elif labeling_type == LabelTypes.NO_LABEL.value:
52        return 0
53    else:
54        raise Exception('Label not set please select one of LabelTypes(Enum) values!')
55
56
57
58
59def next_event_name(trace: list, prefix_length: int):
60    """Return the event event_name at prefix length or 0 if out of range.
61
62    """
63    if prefix_length < len(trace):
64        next_event = trace[prefix_length]
65        name = next_event['concept:name']
66        return name
67    else:
68        return 0