Merge pull request #42 from AutomatedProcessImprovement/output-expansion

Output expansion v1

Author: iharsuvorau

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "wta"
-version = "1.3.6"
+version = "1.3.7"
 description = "Waiting time analysis of activity transitions in event logs"
 authors = ["Ihar Suvorau <ihar.suvorau@gmail.com>", "David Chapela de la Campa <david.chapela@ut.ee>"]
 license = "Apache-2.0"

src/wta/activity_transitions.py

@@ -1,6 +1,6 @@
 import concurrent.futures
 import multiprocessing
-from typing import Dict, Optional, List
+from typing import Dict, Optional, Union
 
 import click
 import numpy as np
@@ -11,252 +11,107 @@
 from wta.helpers import print_section_boundaries, convert_timestamp_columns_to_datetime, log_ids_non_nil, \
     EventLogIDs, TRANSITION_COLUMN_KEY
 from wta.waiting_time import analysis as wt_analysis
+from wta.calendars.calendars import make as make_calendar
+
+
+CONVERT_COLUMNS = ['wt_total', 'wt_contention', 'wt_batching', 'wt_prioritization', 'wt_unavailability', 'wt_extraneous']
+ORDERED_COLUMNS = [
+    'start_time',
+    'end_time',
+    'source_activity',
+    'source_resource',
+    'destination_activity',
+    'destination_resource',
+    'case_id',
+    'wt_total',
+    'wt_contention',
+    'wt_batching',
+    'wt_prioritization',
+    'wt_unavailability',
+    'wt_extraneous'
+]
 
 
 @print_section_boundaries('Activity Transitions Analysis')
-def identify(
-        log: pd.DataFrame,
-        parallel_activities: Dict[str, set],
-        parallel_run=True,
-        log_ids: Optional[EventLogIDs] = None,
-        calendar: Optional[Dict] = None,
-        group_results: bool = True) -> Optional[pd.DataFrame]:
-    from wta.calendars.calendars import make as make_calendar
-
+def identify(log: pd.DataFrame, parallel_activities: Dict[str, set], parallel_run: bool = True,
+             log_ids: Optional[EventLogIDs] = None, calendar: Optional[Dict] = None) -> Optional[pd.DataFrame]:
     click.echo(f'Parallel run: {parallel_run}')
-
     log_ids = log_ids_non_nil(log_ids)
+    log_calendar = make_calendar_if_none(log, log_ids, calendar)
+    run_func = __multiprocess_run if parallel_run else __sequential_run
+    all_items = run_func(log, log_ids, log_calendar, parallel_activities)
+    return None if len(all_items) == 0 else process_all_items(all_items)
 
-    if not calendar:
-        log_calendar = make_calendar(log, granularity=GRANULARITY_MINUTES, log_ids=log_ids)
-    else:
-        log_calendar = calendar
 
-    if parallel_run:
-        all_items = __multiprocess_run(log, log_ids, log_calendar, parallel_activities)
-    else:
-        all_items = __sequential_run(log, log_ids, log_calendar, parallel_activities)
+def process_all_items(all_items: pd.DataFrame) -> pd.DataFrame:
+    # Convert time columns to seconds
+    columns_to_convert = [col for col in CONVERT_COLUMNS if col in all_items.columns]
+    all_items[columns_to_convert] = all_items[columns_to_convert].applymap(lambda x: pd.to_timedelta(x).total_seconds())
 
-    if len(all_items) == 0:
-        return None
+    # Return the dataframe in the order of ORDERED_COLUMNS
+    return all_items[ORDERED_COLUMNS]
 
-    if group_results:
-        result = __join_per_case_items(all_items, log_ids=log_ids)
-        if result is not None:
-            result['wt_total_seconds'] = result[log_ids.wt_total] / np.timedelta64(1, 's')
-    else:
-        result = __create_single_dataframe(all_items)
 
-    return result
+def make_calendar_if_none(log, log_ids, calendar):
+    return make_calendar(log, granularity=GRANULARITY_MINUTES, log_ids=log_ids) if not calendar else calendar
 
 
 def __sequential_run(log, log_ids, calendar, parallel_activities):
-    results = []
     log_grouped = log.groupby(by=log_ids.case)
-
-    for (case_id, case) in log_grouped:
-        case = case.sort_values(by=[log_ids.end_time, log_ids.start_time])
-        result = __identify_transitions_per_case_and_make_report(
-            case,
-            parallel_activities=parallel_activities,
-            case_id=case_id,
-            log_calendar=calendar,
-            log=log,
-            log_ids=log_ids)
-
-        if result is not None:
-            results.append(result)
-
-    return results
+    results_transitions = [identify_transitions_and_report(sort_case(case, log_ids), parallel_activities, case_id, calendar, log, log_ids)
+                           for case_id, case in log_grouped]
+    return concatenate_transitions_if_exists(results_transitions)
 
 
 def __multiprocess_run(log, log_ids, calendar, parallel_activities):
-    all_items = []
     n_cores = multiprocessing.cpu_count() - 1
     handles = []
     log_grouped = log.groupby(by=log_ids.case)
 
     with concurrent.futures.ProcessPoolExecutor(max_workers=n_cores) as executor:
-        for (case_id, case) in tqdm(log_grouped, desc='Submitting tasks for concurrent execution'):
-            case = case.sort_values(by=[log_ids.end_time, log_ids.start_time])
-            handle = executor.submit(__identify_transitions_per_case_and_make_report,
-                                     case,
-                                     parallel_activities=parallel_activities,
-                                     case_id=case_id,
-                                     log_calendar=calendar,
-                                     log=log,
-                                     log_ids=log_ids)
-            handles.append(handle)
-
-    for h in tqdm(handles, desc='Waiting for tasks to finish'):
-        done = h.done()
-        result = h.result()
-        if done and not result.empty:
-            all_items.append(result)
+        handles = [submit_task(executor, sort_case(case, log_ids), parallel_activities, case_id, calendar, log, log_ids)
+                   for case_id, case in tqdm(log_grouped, desc='Submitting tasks for concurrent execution')]
 
-    return all_items
+    all_transitions = [h.result() for h in tqdm(handles, desc='Waiting for tasks to finish') if not h.result().empty]
+    return concatenate_transitions_if_exists(all_transitions)
 
 
-def __identify_transitions_per_case_and_make_report(case: pd.DataFrame, **kwargs) -> pd.DataFrame:
-    parallel_activities = kwargs['parallel_activities']
-    case_id = kwargs['case_id']
-    log_calendar = kwargs['log_calendar']
-    log = kwargs['log']
-    log_ids = log_ids_non_nil(kwargs.get('log_ids'))
+def sort_case(case, log_ids):
+    return case.sort_values(by=[log_ids.end_time, log_ids.start_time])
 
-    case = case.sort_values(by=[log_ids.end_time, log_ids.start_time]).copy()
 
-    # converting timestamps to datetime
-    log = convert_timestamp_columns_to_datetime(log, log_ids)
-    case = convert_timestamp_columns_to_datetime(case, log_ids)
-
-    __mark_activity_transitions(case, parallel_activities, log_ids=log_ids)
-
-    transitions = wt_analysis.run(case, log_calendar, log, log_ids=log_ids)
+def submit_task(executor, case, parallel_activities, case_id, calendar, log, log_ids):
+    return executor.submit(identify_transitions_and_report, case, parallel_activities, case_id, calendar, log, log_ids)
 
-    transitions_with_frequency = __calculate_frequency_and_duration(transitions, log_ids=log_ids)
 
-    # dropping edge cases with Start and End as an activity
-    starts_ends_values = ['Start', 'End']
-    starts_and_ends = (transitions_with_frequency['source_activity'].isin(starts_ends_values)
-                       & transitions_with_frequency['source_resource'].isin(starts_ends_values)) \
-                      | (transitions_with_frequency['destination_activity'].isin(starts_ends_values)
-                         & transitions_with_frequency['destination_resource'].isin(starts_ends_values))
-    transitions_with_frequency = transitions_with_frequency[starts_and_ends == False]
+def concatenate_transitions_if_exists(results_transitions):
+    return pd.concat(results_transitions, ignore_index=True) if results_transitions else None
 
-    # attaching case ID as additional information
-    transitions_with_frequency['case_id'] = case_id
 
-    return transitions_with_frequency
-
-
-def __mark_activity_transitions(
-        case: pd.DataFrame,
-        parallel_activities: Optional[Dict[str, set]] = None,
-        log_ids: Optional[EventLogIDs] = None):
-    log_ids = log_ids_non_nil(log_ids)
-
-    # NOTE: we assume (a) the case was sorted by end time
+def identify_transitions_and_report(case, parallel_activities, case_id, log_calendar, log, log_ids):
+    case = convert_timestamp_columns_to_datetime(case, log_ids)
+    log = convert_timestamp_columns_to_datetime(log, log_ids)
+    mark_activity_transitions(case, parallel_activities, log_ids=log_ids)
+    transitions = wt_analysis.run(case, log_calendar, log, log_ids=log_ids)
+    transitions['case_id'] = case_id
+    return transitions
 
-    if not parallel_activities:
-        parallel_activities = {}
 
+def mark_activity_transitions(case, parallel_activities, log_ids):
     case[log_ids.transition_source_index] = np.NAN
-
-    # processing the case backwards
     reversed_index = list(reversed(case.index))
     for i in range(len(reversed_index)):
         non_concurrent_previous_event_found = False
-
         index = reversed_index[i]
         current_event = case.loc[index]
         parallel_activities_for_current_event = parallel_activities.get(current_event[log_ids.activity], [])
-
         previous_event_index_delta = i + 1
-        while not non_concurrent_previous_event_found:
-            if previous_event_index_delta > len(reversed_index) - 1:
-                break
-
+        while not non_concurrent_previous_event_found and previous_event_index_delta <= len(reversed_index) - 1:
             previous_event_index = reversed_index[previous_event_index_delta]
             previous_event = case.loc[previous_event_index]
-            # Check if they are overlapping
             overlapping_activity_instances = previous_event[log_ids.end_time] > current_event[log_ids.start_time]
-            if previous_event[
-                log_ids.activity] in parallel_activities_for_current_event or overlapping_activity_instances:
-                # If they are concurrent activities, or overlapping instances, jump to the previous event
+            if previous_event[log_ids.activity] in parallel_activities_for_current_event or overlapping_activity_instances:
                 previous_event_index_delta += 1
             else:
-                # If they are not concurrent nor overlapping, transition!
                 case.at[index, TRANSITION_COLUMN_KEY] = previous_event_index
                 non_concurrent_previous_event_found = True
-
-
-def __calculate_frequency_and_duration(transitions: pd.DataFrame,
-                                       log_ids: Optional[EventLogIDs] = None) -> pd.DataFrame:
-    log_ids = log_ids_non_nil(log_ids)
-
-    # calculating frequency per case of the transitions with the same activities and resources
-    columns = transitions.columns.tolist()
-    transition_with_frequency = pd.DataFrame(columns=columns)
-    for (pair, records) in transitions.groupby(by=['source_activity', 'source_resource',
-                                                   'destination_activity', 'destination_resource']):
-        transition_with_frequency = pd.concat([transition_with_frequency, pd.DataFrame({
-            'source_activity': [pair[0]],
-            'source_resource': [pair[1]],
-            'destination_activity': [pair[2]],
-            'destination_resource': [pair[3]],
-            'frequency': [len(records)],
-            'transition_type': [records['transition_type'].iloc[0]],
-            log_ids.wt_total: [records[log_ids.wt_total].sum()],
-            log_ids.wt_batching: [records[log_ids.wt_batching].sum()],
-            log_ids.wt_prioritization: [records[log_ids.wt_prioritization].sum()],
-            log_ids.wt_contention: [records[log_ids.wt_contention].sum()],
-            log_ids.wt_unavailability: [records[log_ids.wt_unavailability].sum()],
-            log_ids.wt_extraneous: [records[log_ids.wt_extraneous].sum()],
-        })], ignore_index=True)
-    return transition_with_frequency
-
-
-def __create_single_dataframe(items: List[pd.DataFrame]) -> Optional[pd.DataFrame]:
-    items = list(filter(lambda df: not df.empty, items))
-    if len(items) == 0:
-        return None
-    else:
-        return pd.concat(items).reset_index(drop=True)
-
-
-def __join_per_case_items(items: List[pd.DataFrame], log_ids: Optional[EventLogIDs] = None) -> Optional[pd.DataFrame]:
-    """Joins a list of items summing up frequency and duration."""
-
-    log_ids = log_ids_non_nil(log_ids)
-
-    items = list(filter(lambda df: not df.empty, items))
-
-    if len(items) == 0:
-        return None
-
-    columns = ['source_activity', 'source_resource', 'destination_activity', 'destination_resource']
-    grouped = pd.concat(items).groupby(columns)
-    result = pd.DataFrame(columns=columns)
-    for pair_index, group in grouped:
-        source_activity, source_resource, destination_activity, destination_resource = pair_index
-        group_wt_total: pd.Timedelta = group[log_ids.wt_total].sum()
-
-        group_wt_batching = pd.Timedelta(0)
-        if log_ids.wt_batching in group.columns:
-            group_wt_batching = group[log_ids.wt_batching].sum()
-
-        group_wt_prioritization = pd.Timedelta(0)
-        if log_ids.wt_prioritization in group.columns:
-            group_wt_prioritization = group[log_ids.wt_prioritization].sum()
-
-        group_wt_contention = pd.Timedelta(0)
-        if log_ids.wt_contention in group.columns:
-            group_wt_contention = pd.to_timedelta(group[log_ids.wt_contention]).sum()
-
-        group_wt_unavailability = pd.Timedelta(0)
-        if log_ids.wt_unavailability in group.columns:
-            group_wt_unavailability = pd.to_timedelta(group[log_ids.wt_unavailability]).sum()
-
-        group_wt_extraneous = pd.Timedelta(0)
-        if log_ids.wt_extraneous in group.columns:
-            group_wt_extraneous = pd.to_timedelta(group[log_ids.wt_extraneous]).sum()
-
-        group_frequency: float = group['frequency'].sum()
-        group_case_id: str = ','.join(group['case_id'].astype(str).unique())
-        result = pd.concat([result, pd.DataFrame({
-            'source_activity': [source_activity],
-            'source_resource': [source_resource],
-            'destination_activity': [destination_activity],
-            'destination_resource': [destination_resource],
-            'frequency': [group_frequency],
-            'cases': [group_case_id],
-            log_ids.wt_total: [group_wt_total],
-            log_ids.wt_batching: [group_wt_batching],
-            log_ids.wt_prioritization: [group_wt_prioritization],
-            log_ids.wt_contention: [group_wt_contention],
-            log_ids.wt_unavailability: [group_wt_unavailability],
-            log_ids.wt_extraneous: [group_wt_extraneous]
-        })], ignore_index=True)
-    result.reset_index(drop=True, inplace=True)
-    return result

src/wta/cli.py

@@ -55,7 +55,7 @@ def _run(
     csv_path = output_path.with_suffix('.csv')
 
     print(f'Saving transitions report to {csv_path}')
-    report.transitions_report.to_csv(csv_path, index=False)
+    report.to_csv(csv_path, index=False)
 
     json_path = output_path.with_suffix('.json')
 

src/wta/main.py

@@ -47,13 +47,10 @@ def run(log_path: Path,
     log[log_ids.wt_total] = log[log_ids.start_time] - log[log_ids.enabled_time]
 
     parallel_activities = parallel_activities_with_heuristic_oracle(log, log_ids=log_ids)
-    transitions_data = activity_transitions.identify(
-        log, parallel_activities, parallel_run, log_ids=log_ids, calendar=calendar, group_results=group_results)
+    transitions_data = activity_transitions.identify(log, parallel_activities, parallel_run, log_ids=log_ids,
+                                                     calendar=calendar)
 
-    if group_results:
-        return TransitionsReport(transitions_data, log, log_ids)
-    else:
-        return transitions_data
+    return transitions_data
 
 
 @print_section_boundaries('Batch Analysis')

src/wta/waiting_time/analysis.py

@@ -63,15 +63,15 @@ def run(case: pd.DataFrame,
             wt_unavailability = pd.Timedelta(0)
             wt_extraneous = pd.Timedelta(0)
 
-        transition_type = 'transition'
 
         # appending the handoff data
         transition = pd.DataFrame({
             'source_activity': [source[log_ids.activity]],
             'source_resource': [source[log_ids.resource]],
             'destination_activity': [destination[log_ids.activity]],
             'destination_resource': [destination[log_ids.resource]],
-            'transition_type': [transition_type],
+            'start_time': [source[log_ids.start_time]],
+            'end_time': [source[log_ids.end_time]],
             log_ids.wt_total: [wt_total],
             log_ids.wt_batching: [wt_batching],
             log_ids.wt_prioritization: [wt_prioritization],

tests/test_core.py

@@ -40,8 +40,3 @@ def handoffs(assets_path) -> List[pd.DataFrame]:
         pd.read_csv(assets_path / 'bimp-example_case_handoff_5.csv'),
         pd.read_csv(assets_path / 'bimp-example_case_handoff_6.csv'),
     ]
-
-
-def test_join_handoffs(handoffs):
-    result = wta.activity_transitions.__join_per_case_items(handoffs)
-    assert result is not None and not result.empty