Skip to content

Loading Event Logs

ergminer.read_csv(filepath, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp', resource_col=None, remove_incomplete_traces=True, min_events_per_trace=2)

Load an event log from a CSV file.

Parameters:

Name Type Description Default
filepath str

Path to the CSV file.

 required
case_id_col str

Column name for case identifiers.

 'case_id'
activity_col str

Column name for activity names.

 'activity_name'
timestamp_col str

Column name for timestamps.

 'timestamp'
resource_col Optional[str]

Column name for resource identifiers (optional).

 None
remove_incomplete_traces bool

Drop incomplete/partial traces before returning.

 True
min_events_per_trace int

Minimum events a trace must have to be kept.

 2

Returns:

Type Description
DataFrame

pd.DataFrame with the caller's original column names.
Source code in src\ergminer\read.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def read_csv(
    filepath: str,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
    resource_col: Optional[str] = None,
    remove_incomplete_traces: bool = True,
    min_events_per_trace: int = 2,
) -> pd.DataFrame:
    """Load an event log from a CSV file.

    Args:
        filepath:                 Path to the CSV file.
        case_id_col:              Column name for case identifiers.
        activity_col:             Column name for activity names.
        timestamp_col:            Column name for timestamps.
        resource_col:             Column name for resource identifiers (optional).
        remove_incomplete_traces: Drop incomplete/partial traces before returning.
        min_events_per_trace:     Minimum events a trace must have to be kept.

    Returns:
        ``pd.DataFrame`` with the caller's original column names.
    """
    from .event_log_loader import EventLogLoader

    loader = EventLogLoader()
    loader.load_csv(filepath, case_id_col, activity_col, timestamp_col, resource_col)
    if remove_incomplete_traces:
        loader.remove_incomplete_and_partial_traces(min_events_per_trace)
    return loader.log_df.copy()

ergminer.read_xes(filepath, case_id_col='case:concept:name', activity_col='concept:name', timestamp_col='time:timestamp', resource_col='org:resource', remove_incomplete_traces=True, min_events_per_trace=2)

Load an event log from an XES file (requires pm4py).

Default column names match the XES / pm4py standard, so no column arguments are usually needed beyond the file path.

Parameters:

Name Type Description Default
filepath str

Path to the .xes file.

 required
case_id_col str

Column name for case identifiers in the XES DataFrame.

 'case:concept:name'
activity_col str

Column name for activity names.

 'concept:name'
timestamp_col str

Column name for timestamps.

 'time:timestamp'
resource_col Optional[str]

Column name for resource identifiers (optional).

 'org:resource'
remove_incomplete_traces bool

Drop incomplete/partial traces before returning.

 True
min_events_per_trace int

Minimum events a trace must have to be kept.

 2

Returns:

Type Description
DataFrame

pd.DataFrame with the caller's original XES column names.

Raises:

Type Description
ImportError

If pm4py is not installed.
Source code in src\ergminer\read.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def read_xes(
    filepath: str,
    case_id_col: str = 'case:concept:name',
    activity_col: str = 'concept:name',
    timestamp_col: str = 'time:timestamp',
    resource_col: Optional[str] = 'org:resource',
    remove_incomplete_traces: bool = True,
    min_events_per_trace: int = 2,
) -> pd.DataFrame:
    """Load an event log from an XES file (requires pm4py).

    Default column names match the XES / pm4py standard, so no column
    arguments are usually needed beyond the file path.

    Args:
        filepath:                 Path to the ``.xes`` file.
        case_id_col:              Column name for case identifiers in the XES DataFrame.
        activity_col:             Column name for activity names.
        timestamp_col:            Column name for timestamps.
        resource_col:             Column name for resource identifiers (optional).
        remove_incomplete_traces: Drop incomplete/partial traces before returning.
        min_events_per_trace:     Minimum events a trace must have to be kept.

    Returns:
        ``pd.DataFrame`` with the caller's original XES column names.

    Raises:
        ImportError: If pm4py is not installed.
    """
    try:
        import pm4py
    except ImportError as exc:
        raise ImportError(
            "read_xes() requires pm4py. Install with: pip install pm4py"
        ) from exc

    log = pm4py.read_xes(str(filepath))
    df = pm4py.convert_to_dataframe(log)
    return read_dataframe(
        df,
        case_id_col=case_id_col,
        activity_col=activity_col,
        timestamp_col=timestamp_col,
        resource_col=resource_col,
        remove_incomplete_traces=remove_incomplete_traces,
        min_events_per_trace=min_events_per_trace,
    )

ergminer.read_dataframe(dataframe, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp', resource_col=None, remove_incomplete_traces=True, min_events_per_trace=2)

Load an event log from an existing pd.DataFrame.

Applies the same timestamp parsing and optional trace filtering as read_csv().

Parameters:

Name Type Description Default
dataframe DataFrame

Source DataFrame.

 required
case_id_col str

Column name for case identifiers.

 'case_id'
activity_col str

Column name for activity names.

 'activity_name'
timestamp_col str

Column name for timestamps.

 'timestamp'
resource_col Optional[str]

Column name for resource identifiers (optional).

 None
remove_incomplete_traces bool

Drop incomplete/partial traces before returning.

 True
min_events_per_trace int

Minimum events a trace must have to be kept.

 2

Returns:

Type Description
DataFrame

Preprocessed pd.DataFrame with the caller's original column names.
Source code in src\ergminer\read.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def read_dataframe(
    dataframe: pd.DataFrame,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
    resource_col: Optional[str] = None,
    remove_incomplete_traces: bool = True,
    min_events_per_trace: int = 2,
) -> pd.DataFrame:
    """Load an event log from an existing ``pd.DataFrame``.

    Applies the same timestamp parsing and optional trace filtering as
    ``read_csv()``.

    Args:
        dataframe:                Source DataFrame.
        case_id_col:              Column name for case identifiers.
        activity_col:             Column name for activity names.
        timestamp_col:            Column name for timestamps.
        resource_col:             Column name for resource identifiers (optional).
        remove_incomplete_traces: Drop incomplete/partial traces before returning.
        min_events_per_trace:     Minimum events a trace must have to be kept.

    Returns:
        Preprocessed ``pd.DataFrame`` with the caller's original column names.
    """
    from .event_log_loader import EventLogLoader

    loader = EventLogLoader()
    loader.load_dataframe(dataframe, case_id_col, activity_col, timestamp_col, resource_col)
    if remove_incomplete_traces:
        loader.remove_incomplete_and_partial_traces(min_events_per_trace)
    return loader.log_df.copy()