ergminer.filtering

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

Keep only cases whose first event is one of activities.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log DataFrame.	required
activities	Union[str, List[str]]	Activity name or list of activity names to keep.	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
DataFrame	Filtered DataFrame (same columns, subset of cases).

Source code in src\ergminer\filtering.py

def filter_start_activities(
    log: pd.DataFrame,
    activities: Union[str, List[str]],
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> pd.DataFrame:
    """Keep only cases whose first event is one of *activities*.

    Args:
        log:          Event log DataFrame.
        activities:   Activity name or list of activity names to keep.
        case_id_col:  Column name for case identifiers.
        activity_col: Column name for activity names.
        timestamp_col: Column name for timestamps.

    Returns:
        Filtered DataFrame (same columns, subset of cases).
    """
    if isinstance(activities, str):
        activities = [activities]
    activities = set(activities)

    first = (
        log.sort_values(timestamp_col)
           .groupby(case_id_col, sort=False)[activity_col]
           .first()
    )
    keep = first[first.isin(activities)].index
    return log[log[case_id_col].isin(keep)].copy()

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

Keep only cases whose last event is one of activities.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log DataFrame.	required
activities	Union[str, List[str]]	Activity name or list of activity names to keep.	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
DataFrame	Filtered DataFrame (same columns, subset of cases).

Source code in src\ergminer\filtering.py

def filter_end_activities(
    log: pd.DataFrame,
    activities: Union[str, List[str]],
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> pd.DataFrame:
    """Keep only cases whose last event is one of *activities*.

    Args:
        log:          Event log DataFrame.
        activities:   Activity name or list of activity names to keep.
        case_id_col:  Column name for case identifiers.
        activity_col: Column name for activity names.
        timestamp_col: Column name for timestamps.

    Returns:
        Filtered DataFrame (same columns, subset of cases).
    """
    if isinstance(activities, str):
        activities = [activities]
    activities = set(activities)

    last = (
        log.sort_values(timestamp_col)
           .groupby(case_id_col, sort=False)[activity_col]
           .last()
    )
    keep = last[last.isin(activities)].index
    return log[log[case_id_col].isin(keep)].copy()

ergminer.filter_case_size(log, min_size=1, max_size=None, case_id_col='case_id')

Keep only cases whose event count is within [min_size, max_size].

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log DataFrame.	required
min_size	int	Minimum number of events a case must have (inclusive).	1
max_size	Optional[int]	Maximum number of events a case may have (inclusive). None = no upper limit.	None
case_id_col	str	Column name for case identifiers.	'case_id'

Returns:

Type	Description
DataFrame	Filtered DataFrame (same columns, subset of cases).

Source code in src\ergminer\filtering.py

def filter_case_size(
    log: pd.DataFrame,
    min_size: int = 1,
    max_size: Optional[int] = None,
    case_id_col: str = 'case_id',
) -> pd.DataFrame:
    """Keep only cases whose event count is within [*min_size*, *max_size*].

    Args:
        log:         Event log DataFrame.
        min_size:    Minimum number of events a case must have (inclusive).
        max_size:    Maximum number of events a case may have (inclusive). ``None`` = no upper limit.
        case_id_col: Column name for case identifiers.

    Returns:
        Filtered DataFrame (same columns, subset of cases).
    """
    sizes = log.groupby(case_id_col, sort=False).size()
    mask = sizes >= min_size
    if max_size is not None:
        mask &= sizes <= max_size
    keep = sizes[mask].index
    return log[log[case_id_col].isin(keep)].copy()

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

Keep only cases whose activity sequence matches one of variants.

Each element of variants may be either a tuple of activity strings or a '->'-joined string (e.g. 'A -> B -> C'), which is normalised internally to a tuple.

Parameters:

Name	Type	Description	Default
log	DataFrame	Event log DataFrame.	required
variants	Union[List[Tuple[str, ...]], List[str]]	List of variants to keep (tuples or arrow-separated strings).	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
DataFrame	Filtered DataFrame (same columns, subset of cases).

Source code in src\ergminer\filtering.py

def filter_variants(
    log: pd.DataFrame,
    variants: Union[List[Tuple[str, ...]], List[str]],
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
) -> pd.DataFrame:
    """Keep only cases whose activity sequence matches one of *variants*.

    Each element of *variants* may be either a ``tuple`` of activity strings or
    a ``'->'``-joined string (e.g. ``'A -> B -> C'``), which is normalised
    internally to a tuple.

    Args:
        log:          Event log DataFrame.
        variants:     List of variants to keep (tuples or arrow-separated strings).
        case_id_col:  Column name for case identifiers.
        activity_col: Column name for activity names.
        timestamp_col: Column name for timestamps.

    Returns:
        Filtered DataFrame (same columns, subset of cases).
    """
    def _normalise(v: Union[Tuple[str, ...], str]) -> Tuple[str, ...]:
        if isinstance(v, str):
            return tuple(s.strip() for s in v.split('->'))
        return tuple(v)

    target = {_normalise(v) for v in variants}

    keep = []
    for case_id, case_df in log.groupby(case_id_col, sort=False):
        seq = tuple(case_df.sort_values(timestamp_col)[activity_col].tolist())
        if seq in target:
            keep.append(case_id)

    return log[log[case_id_col].isin(keep)].copy()