Время прочтения: 2 мин.
Давайте поподробнее. На вход алгоритму подается лог файл, содержащий 3 столбца (id_column, dt_column, activity_column), который мы преобразовываем в pandas Data Frame. Далее, для более удобного восприятия информации, нам необходимо максимально подробно и правильно поставить задачу: «В результате алгоритма мы хотим видеть массив данных, содержащий только те цепочки событий графа процесса, в которых обязательно есть либо интересующий нас переход между шагами процесса, либо определенный шаг, либо и то и другое условие вместе.» Для этого напишем функцию с несколькими входными параметрами: исходный датафрейм, список ребер графа, список вершин графа.
def rules_cheking(frame, edges=None, activity=None):
Для начала создадим новый датафрейм из исходного, в котором будет новый столбец ‘next_activity’, который мы заполним таким образом, чтобы там был каждое следующее за исходным действие в рамках одного идентификатора. Образовавшиеся в этом столбце значения ‘nan’ заполним строкой ‘end’, это значит, что шагов после этого нет.
rules_check = pd.DataFrame(columns=['id_column', 'activity_column', 'dt_column', 'next_activity'])
rules_check[['id_column', 'activity_column', 'dt_column']] = frame[['id_column', 'activity_column', 'dt_column']]
rules_check['next_activity'] = frame.groupby('id_column')['activity_column'].shift(-1)
rules_check['next_activity'].fillna('end', inplace=True)
После подготовки нового фрейма нужно прописать условия, которые будут определять, какие именно вы задали правила для данных. Первое условие будет выполняться, если вас интересуют только некоторые шаги процесса — activity.
if (activity != None) & (edges == None) :
id_rules = rules_check.loc[rules_check['activity_column'].isin(activity)]['id_column']
rules_check = frame[frame['id_column'].isin(id_rules)]
Второе условие будет выполняться, если вас интересуют только некоторые переходы — edges.
elif (activity == None) & (edges != None) :
id_rules = rules_check.loc[rules_check[['activity_column', 'next_activity']].apply(tuple, 1).isin(
pd.DataFrame(edges, columns=['activity_column', 'next_activity']).apply(tuple, 1))]['id_column']
rules_check = frame[frame['id_column'].isin(id_rules)]
Иначе поиск будет выполняться по обоим заданным условиям одновременно.
else:
id_rules_edges = rules_check.loc[rules_check[['activity_column', 'next_activity']].apply(tuple, 1).isin(
pd.DataFrame(edges, columns=['activity_column', 'next_activity']).apply(tuple, 1))]['id_column']
rules_check_edges = frame[frame['id_column'].isin(id_rules_edges)]
id_rules_all = rules_check.loc[rules_check['activity_column'].isin(activity)]['id_column']
rules_check = frame[frame['id_column'].isin(id_rules_all)]
Возвращаем датафрейм, в котором содержатся только те цепочки событий, которые удовлетворяют заданным вами правилам.
return rules_check
Полученный набор данных вы можете дальше анализировать другими алгоритмами Process Mining, например, найти временные аномалии или построить граф, с учетом ваших правил.