ergminer.utils

ergminer.format_dataframe(dataframe, case_id_col, activity_col, timestamp_col, resource_col=None)

Return a copy of dataframe with columns renamed to ergminer standard names.

Standard names: case_id, activity_name, timestamp, resource_id. Only the columns listed are renamed; all other columns are preserved as-is.

Parameters:

Name	Type	Description	Default
dataframe	DataFrame	Source event log DataFrame.	required
case_id_col	str	Column name for case identifiers.	required
activity_col	str	Column name for activity names.	required
timestamp_col	str	Column name for timestamps.	required
resource_col	Optional[str]	Column name for resource identifiers (optional).	None

Returns:

Type	Description
DataFrame	New DataFrame with standardised column names.

Source code in src\ergminer\utils.py

def format_dataframe(
    dataframe: pd.DataFrame,
    case_id_col: str,
    activity_col: str,
    timestamp_col: str,
    resource_col: Optional[str] = None,
) -> pd.DataFrame:
    """Return a copy of *dataframe* with columns renamed to ergminer standard names.

    Standard names: ``case_id``, ``activity_name``, ``timestamp``, ``resource_id``.
    Only the columns listed are renamed; all other columns are preserved as-is.

    Args:
        dataframe:     Source event log 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).

    Returns:
        New DataFrame with standardised column names.
    """
    rename_map = {
        case_id_col:   'case_id',
        activity_col:  'activity_name',
        timestamp_col: 'timestamp',
    }
    if resource_col and resource_col in dataframe.columns:
        rename_map[resource_col] = 'resource_id'
    return dataframe.rename(columns=rename_map)

ergminer.get_start_activities(log, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp')

Return {activity: count} for the first event in each case.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log 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 (used to determine first event).	'timestamp'

Returns:

Type	Description
Dict[str, int]	Dict mapping each start-activity name to the number of cases it starts.

Source code in src\ergminer\utils.py

def get_start_activities(
    log: pd.DataFrame,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> Dict[str, int]:
    """Return ``{activity: count}`` for the first event in each case.

    Args:
        log:           Event log DataFrame.
        case_id_col:   Column name for case identifiers.
        activity_col:  Column name for activity names.
        timestamp_col: Column name for timestamps (used to determine first event).

    Returns:
        Dict mapping each start-activity name to the number of cases it starts.
    """
    first = (
        log.sort_values(timestamp_col)
           .groupby(case_id_col, sort=False)[activity_col]
           .first()
    )
    return dict(Counter(first))

ergminer.get_end_activities(log, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp')

Return {activity: count} for the last event in each case.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log 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 (used to determine last event).	'timestamp'

Returns:

Type	Description
Dict[str, int]	Dict mapping each end-activity name to the number of cases it ends.

Source code in src\ergminer\utils.py

def get_end_activities(
    log: pd.DataFrame,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> Dict[str, int]:
    """Return ``{activity: count}`` for the last event in each case.

    Args:
        log:           Event log DataFrame.
        case_id_col:   Column name for case identifiers.
        activity_col:  Column name for activity names.
        timestamp_col: Column name for timestamps (used to determine last event).

    Returns:
        Dict mapping each end-activity name to the number of cases it ends.
    """
    last = (
        log.sort_values(timestamp_col)
           .groupby(case_id_col, sort=False)[activity_col]
           .last()
    )
    return dict(Counter(last))

ergminer.get_variants(log, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp')

Return {variant_tuple: count} for all cases in log.

A variant is the ordered sequence (tuple) of activity names within a case, sorted by timestamp_col.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log 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'

Returns:

Type	Description
Dict[Tuple[str, ...], int]	Dict mapping each variant (tuple of activity strings) to its frequency.

Source code in src\ergminer\utils.py

def get_variants(
    log: pd.DataFrame,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> Dict[Tuple[str, ...], int]:
    """Return ``{variant_tuple: count}`` for all cases in *log*.

    A variant is the ordered sequence (tuple) of activity names within a case,
    sorted by *timestamp_col*.

    Args:
        log:           Event log DataFrame.
        case_id_col:   Column name for case identifiers.
        activity_col:  Column name for activity names.
        timestamp_col: Column name for timestamps.

    Returns:
        Dict mapping each variant (tuple of activity strings) to its frequency.
    """
    variants: Counter = Counter()
    for _, case_df in log.groupby(case_id_col, sort=False):
        seq = tuple(case_df.sort_values(timestamp_col)[activity_col].tolist())
        variants[seq] += 1
    return dict(variants)

ergminer.get_activity_labels(log, activity_col='activity_name')

Return a sorted list of unique activity names present in log.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log DataFrame.	required
activity_col	str	Column name for activity names.	'activity_name'

Returns:

Type	Description
List[str]	Sorted list of unique activity label strings.

Source code in src\ergminer\utils.py

def get_activity_labels(
    log: pd.DataFrame,
    activity_col: str = 'activity_name',
) -> List[str]:
    """Return a sorted list of unique activity names present in *log*.

    Args:
        log:          Event log DataFrame.
        activity_col: Column name for activity names.

    Returns:
        Sorted list of unique activity label strings.
    """
    return sorted(log[activity_col].dropna().unique().tolist())