ergminer.conformance

ergminer.conformance_erg(log, erg, sim_log=None, n_simulations=10, sim_time=5000.0, seed=42, case_id_col='case_id', activity_col='activity_name', timestamp_col='timestamp', resource_col='resource_id', verbose=False)

Run conformance checking between log and erg.

Compares the observed behaviour in log against the ERG model using twelve complementary fitness metrics. If sim_log is not provided, play_out() is called automatically to generate simulation replications.

Parameters:

Name	Type	Description	Default
log	DataFrame	Original event log DataFrame (user column names).	required
erg	'ERG'	The ERG model to check against.	required
sim_log	Optional[DataFrame]	Pre-computed simulation log from play_out() (optional). Must contain _erg_run column. If None, simulations are run automatically.	None
n_simulations	int	Number of simulation replications (used only when sim_log is None).	10
sim_time	float	Simulation horizon passed to play_out() (used only when sim_log is None).	5000.0
seed	int	Base random seed for simulations (used only when sim_log is None).	42
case_id_col	str	Column name for case identifiers in log.	'case_id'
activity_col	str	Column name for activity names in log.	'activity_name'
timestamp_col	str	Column name for timestamps in log.	'timestamp'
resource_col	str	Column name for resource identifiers in log.	'resource_id'
verbose	bool	Print progress to stdout.	False

Returns:

Type	Description
'ConformanceResult'	ConformanceResult with all check scores and a printable report.

Source code in src\ergminer\conformance.py

def conformance_erg(
    log: pd.DataFrame,
    erg: 'ERG',
    sim_log: Optional[pd.DataFrame] = None,
    n_simulations: int = 10,
    sim_time: float = 5000.0,
    seed: int = 42,
    case_id_col: str = 'case_id',
    activity_col: str = 'activity_name',
    timestamp_col: str = 'timestamp',
    resource_col: str = 'resource_id',
    verbose: bool = False,
) -> 'ConformanceResult':
    """Run conformance checking between *log* and *erg*.

    Compares the observed behaviour in *log* against the ERG model using
    twelve complementary fitness metrics.  If *sim_log* is not provided,
    ``play_out()`` is called automatically to generate simulation replications.

    Args:
        log:           Original event log DataFrame (user column names).
        erg:           The ``ERG`` model to check against.
        sim_log:       Pre-computed simulation log from ``play_out()`` (optional).
                       Must contain ``_erg_run`` column. If ``None``, simulations
                       are run automatically.
        n_simulations: Number of simulation replications (used only when
                       *sim_log* is ``None``).
        sim_time:      Simulation horizon passed to ``play_out()`` (used only
                       when *sim_log* is ``None``).
        seed:          Base random seed for simulations (used only when
                       *sim_log* is ``None``).
        case_id_col:   Column name for case identifiers in *log*.
        activity_col:  Column name for activity names in *log*.
        timestamp_col: Column name for timestamps in *log*.
        resource_col:  Column name for resource identifiers in *log*.
        verbose:       Print progress to stdout.

    Returns:
        ``ConformanceResult`` with all check scores and a printable report.
    """


    # ── 1. Generate sim log if not provided ──────────────────────────────────
    if sim_log is None:
        if verbose:
            print(f"Running {n_simulations} simulation(s) for conformance …")
        sim_log = play_out(
            erg,
            n=n_simulations,
            sim_time=sim_time,
            seed=seed,
            verbose=verbose,
        )
        actual_n = n_simulations
    else:
        actual_n = sim_log['_erg_run'].nunique() if '_erg_run' in sim_log.columns else 1

    # ── 2. Serialise ERG to temp ERGML and parse nodes/arcs ─────────────────
    ergml_xml = erg_to_ergml(erg, sim_time=sim_time)
    tmp_file = tempfile.NamedTemporaryFile(
        suffix='.ergml', delete=False, mode='w', encoding='utf-8'
    )
    try:
        tmp_file.write(ergml_xml)
        tmp_file.close()
        tmp_path = tmp_file.name

        erg_name, nodes, arcs_from, start_events, end_events, res_caps, _ = parse_ergml(tmp_path)
    finally:
        try:
            os.remove(tmp_path)
        except OSError:
            pass

    # ── 3. Rename original log columns to internal standard names ────────────
    # The sim log always uses fixed internal column names (case_id, timestamp,
    # activity_name, resource_id).  Rename the original log to match so both
    # DataFrames use the same column names when passed to ERGConformanceChecker.
    rename_map = {
        case_id_col:   'case_id',
        activity_col:  'activity_name',
        timestamp_col: 'timestamp',
    }
    if resource_col and resource_col in log.columns:
        rename_map[resource_col] = 'resource_id'
    orig_log = log.rename(columns=rename_map).copy()

    # Drop incomplete cases from orig (mirror of conformance_testing behaviour)
    if end_events:
        completed_cases = orig_log.loc[
            orig_log['activity_name'].isin(end_events), 'case_id'
        ].unique()
        orig_log = orig_log[orig_log['case_id'].isin(completed_cases)]

    # ── 4. Build combined sim log with globally unique case IDs ─────────────
    # play_out() restarts case_ids from 1 each run, so prefix them per run.
    combined_parts = []
    run_col = '_erg_run' if '_erg_run' in sim_log.columns else None
    if run_col:
        for run_idx, run_df in sim_log.groupby('_erg_run', sort=True):
            part = run_df.copy()
            part['case_id'] = part['case_id'].astype(str).apply(
                lambda x: f"run{run_idx}_case{x}"
            )
            combined_parts.append(part)
    else:
        combined_parts = [sim_log.copy()]
    combined_sim_log = pd.concat(combined_parts, ignore_index=True)

    # ── 5. Run conformance checker ────────────────────────────────────────────
    if verbose:
        print("Running ERG conformance checks …")

    checker = ERGConformanceChecker(
        original_log  = orig_log,
        sim_log       = combined_sim_log,
        nodes         = nodes,
        arcs_from     = arcs_from,
        case_id_col   = 'case_id',
        activity_col  = 'activity_name',
        timestamp_col = 'timestamp',
        resource_col  = 'resource_id',
    )
    checker.run_all_checks()

    # ── 6. Build ConformanceResult ────────────────────────────────────────────
    summary_df    = checker.summary()
    overall_score = round(float(np.mean([
        r['score'] for r in checker.results.values()
        if isinstance(r.get('score'), (int, float))
    ])), 4)

    group_scores: Dict = {}
    for grp, check_keys in CHECK_GROUPS.items():
        grp_vals = [
            checker.results[k]['score']
            for k in check_keys
            if k in checker.results and isinstance(checker.results[k].get('score'), (int, float))
        ]
        if grp_vals:
            grp_score = round(float(np.mean(grp_vals)), 4)
            grp_pass  = grp_score >= 0.8
            group_scores[grp] = {
                'score':  grp_score,
                'pass':   grp_pass,
                'checks': len(grp_vals),
            }

    result = ConformanceResult()
    result.erg_name        = erg_name or getattr(erg, 'name', 'ERG')
    result.n_simulations   = actual_n
    result.checker_results = checker.results
    result.summary_df      = summary_df
    result.overall_score   = overall_score
    result.group_scores    = group_scores

    return result