Data Classes

DelayConfig

ergminer.DelayConfig dataclass

Tuning parameters for delay analysis. Edit the values to override defaults.

Source code in src\ergminer\erg_miner.py

@dataclass
class DelayConfig:
    """Tuning parameters for delay analysis.  Edit the values to override defaults."""
    immediate_threshold: float = 0.01          # delays below this are instantaneous (seconds)
    immediate_fraction_min: float = 0.02       # min fraction of near-zero delays to be immediate
    resource_blocked_fraction_min: float = 0.5 # min fraction of blocked arcs to confirm immediacy
    min_sample_size: int = 10                  # min observations before fitting a distribution
    min_p_value: float = 0.05                  # min KS p-value to accept a fit
    ks_stat_tolerance: float = 0.05            # tolerance for preferring simpler distributions

ERG

ergminer.ERG

Event Relationship Graph structure.

Source code in src\ergminer\erg_structure.py

class ERG:
    """Event Relationship Graph structure."""

    def __init__(self, name="ERG"):
        """
        Initialize ERG.

        Args:
            name: Name of the ERG
        """
        self.name = name
        self.nodes = {}            # activity name → ERGNode
        self.arcs = []             # list of ERGArc objects
        self.state_variables = {}  # variable name → ERGStateVariable
        self.start_events = set()  # names of start nodes
        self.end_events = set()    # names of end nodes
        self._arcs_from = defaultdict(list)  # source → [ERGArc]
        self._arcs_to = defaultdict(list)    # target → [ERGArc]

    def add_node(self, node):
        """Add a node to the ERG."""
        self.nodes[node.name] = node

    def add_arc(self, arc):
        """Add an arc to the ERG."""
        self.arcs.append(arc)
        self._arcs_from[arc.source].append(arc)
        self._arcs_to[arc.target].append(arc)

    def add_state_variable(self, state_var):
        """Add a state variable to the ERG."""
        self.state_variables[state_var.name] = state_var

    def set_start_event(self, event_name):
        """Mark an event as a start event."""
        self.start_events.add(event_name)

    def set_end_event(self, event_name):
        """Mark an event as an end event."""
        self.end_events.add(event_name)

    def get_node(self, name):
        """Get a node by name."""
        return self.nodes.get(name)

    def get_arcs_from(self, source):
        """Get all arcs originating from a source node."""
        return list(self._arcs_from[source])

    def get_arcs_to(self, target):
        """Get all arcs leading to a target node."""
        return list(self._arcs_to[target])

    def get_arc(self, source, target):
        """Get a specific arc by source and target name."""
        for arc in self.arcs:
            if arc.source == source and arc.target == target:
                return arc
        return None

    def get_statistics(self):
        """Get statistics about the ERG."""
        return {
            'num_nodes': len(self.nodes),
            'num_arcs': len(self.arcs),
            'num_state_variables': len(self.state_variables),
            'num_start_events': len(self.start_events),
            'num_end_events': len(self.end_events),
            'immediate_arcs': sum(1 for arc in self.arcs if arc.is_immediate),
            'delayed_arcs': sum(1 for arc in self.arcs if not arc.is_immediate),
            'guarded_arcs': sum(1 for arc in self.arcs if arc.guard_condition is not None)
        }

    def to_dict(self):
        """Convert ERG to dictionary."""
        return {
            'name': self.name,
            'nodes': [node.to_dict() for node in self.nodes.values()],
            'arcs': [arc.to_dict() for arc in self.arcs],
            'state_variables': [sv.to_dict() for sv in self.state_variables.values()],
            'start_events': list(self.start_events),
            'end_events': list(self.end_events),
            'statistics': self.get_statistics()
        }

    def to_json(self, filepath=None):
        """
        Export ERG to JSON format.

        Args:
            filepath: Optional path to save JSON file

        Returns:
            JSON string representation
        """
        erg_dict = self.to_dict()
        json_str = json.dumps(erg_dict, indent=2)

        if filepath:
            with open(filepath, 'w') as f:
                f.write(json_str)

        return json_str

    def print_summary(self, verbose=True):
        """Print a human-readable summary of the ERG."""
        if not verbose:
            return
        stats = self.get_statistics()

        print(f"\n{'='*60}")
        print(f"Event Relationship Graph: {self.name}")
        print(f"{'='*60}")

        print(f"\nNodes (Events): {stats['num_nodes']}")
        for node in self.nodes.values():
            start_marker = " [START]" if node.name in self.start_events else ""
            end_marker = " [END]" if node.name in self.end_events else ""
            print(f"  - {node.name} ({node.event_type}){start_marker}{end_marker}")
            if node.name not in self.start_events and node.name not in self.end_events:
                if node.state_update_equations:
                    eqs = "; ".join(node.state_update_equations)
                    print(f"      {{ {eqs} }}")
                else:
                    print("      { }")

        print(f"\nState Variables: {stats['num_state_variables']}")
        for sv in self.state_variables.values():
            print(f"  - {sv.name}: {sv.variable_type} for {sv.resource} (init={sv.initial_value})")

        print(f"\nArcs (Transitions): {stats['num_arcs']}")
        print(f"  - Immediate: {stats['immediate_arcs']}")
        print(f"  - Delayed: {stats['delayed_arcs']}")
        print(f"  - Guarded: {stats['guarded_arcs']}")

        print("\nArc Details:")
        for arc in self.arcs:
            guard_str = f" [Guard: {arc.guard_condition}]" if arc.guard_condition else ""

            if arc.is_immediate:
                delay_str = " [Immediate]"
            else:
                delay_str = f" [{arc.delay_distribution}, mean={arc.mean_delay:.3f}]"

            updates_str = ""
            if arc.state_updates:
                updates_str = f" [Updates: {', '.join(arc.state_updates)}]"

            print(f"  {arc.source} -> {arc.target} (p={arc.probability:.3f}){guard_str}{delay_str}{updates_str}")

        print(f"\n{'='*60}\n")

    def generate_graphviz_dot(self, filepath=None):
        """
        Generate GraphViz DOT format for visualization.

        Args:
            filepath: Optional path to save DOT file

        Returns:
            DOT format string
        """
        dot = ["digraph ERG {"]
        dot.append("  rankdir=LR;")
        dot.append("  node [shape=box];")
        dot.append("")

        # Add nodes
        for node in self.nodes.values():
            shape = "ellipse" if node.name in self.start_events or node.name in self.end_events else "box"
            label_parts = [f"{node.name}", f"({node.event_type})"]

            if node.name not in self.start_events and node.name not in self.end_events:
                if node.state_update_equations:
                    inner = "\\n".join(node.state_update_equations)
                    label_parts.append("{" + inner + "}")
                else:
                    label_parts.append("{ }")

            label = "\\n".join(label_parts)
            dot.append(f'  "{node.name}" [label="{label}", shape={shape}];')

        dot.append("")

        # Add arcs
        for arc in self.arcs:
            label_parts = [f"p={arc.probability:.2f}"]

            if arc.guard_condition:
                label_parts.append(f"guard: {arc.guard_condition}")

            if not arc.is_immediate:
                label_parts.append(f"{arc.delay_distribution}")

            label = "\\n".join(label_parts)
            dot.append(f'  "{arc.source}" -> "{arc.target}" [label="{label}"];')

        dot.append("}")
        dot_str = "\n".join(dot)

        if filepath:
            with open(filepath, 'w') as f:
                f.write(dot_str)

        return dot_str

__init__(name='ERG')

Initialize ERG.

Parameters:

Name	Type	Description	Default
name		Name of the ERG	'ERG'

Source code in src\ergminer\erg_structure.py

def __init__(self, name="ERG"):
    """
    Initialize ERG.

    Args:
        name: Name of the ERG
    """
    self.name = name
    self.nodes = {}            # activity name → ERGNode
    self.arcs = []             # list of ERGArc objects
    self.state_variables = {}  # variable name → ERGStateVariable
    self.start_events = set()  # names of start nodes
    self.end_events = set()    # names of end nodes
    self._arcs_from = defaultdict(list)  # source → [ERGArc]
    self._arcs_to = defaultdict(list)    # target → [ERGArc]

add_arc(arc)

Add an arc to the ERG.

Source code in src\ergminer\erg_structure.py

def add_arc(self, arc):
    """Add an arc to the ERG."""
    self.arcs.append(arc)
    self._arcs_from[arc.source].append(arc)
    self._arcs_to[arc.target].append(arc)

add_node(node)

Add a node to the ERG.

Source code in src\ergminer\erg_structure.py

def add_node(self, node):
    """Add a node to the ERG."""
    self.nodes[node.name] = node

add_state_variable(state_var)

Add a state variable to the ERG.

Source code in src\ergminer\erg_structure.py

def add_state_variable(self, state_var):
    """Add a state variable to the ERG."""
    self.state_variables[state_var.name] = state_var

generate_graphviz_dot(filepath=None)

Generate GraphViz DOT format for visualization.

Parameters:

Name	Type	Description	Default
filepath		Optional path to save DOT file	None

Returns:

Type	Description
	DOT format string

Source code in src\ergminer\erg_structure.py

def generate_graphviz_dot(self, filepath=None):
    """
    Generate GraphViz DOT format for visualization.

    Args:
        filepath: Optional path to save DOT file

    Returns:
        DOT format string
    """
    dot = ["digraph ERG {"]
    dot.append("  rankdir=LR;")
    dot.append("  node [shape=box];")
    dot.append("")

    # Add nodes
    for node in self.nodes.values():
        shape = "ellipse" if node.name in self.start_events or node.name in self.end_events else "box"
        label_parts = [f"{node.name}", f"({node.event_type})"]

        if node.name not in self.start_events and node.name not in self.end_events:
            if node.state_update_equations:
                inner = "\\n".join(node.state_update_equations)
                label_parts.append("{" + inner + "}")
            else:
                label_parts.append("{ }")

        label = "\\n".join(label_parts)
        dot.append(f'  "{node.name}" [label="{label}", shape={shape}];')

    dot.append("")

    # Add arcs
    for arc in self.arcs:
        label_parts = [f"p={arc.probability:.2f}"]

        if arc.guard_condition:
            label_parts.append(f"guard: {arc.guard_condition}")

        if not arc.is_immediate:
            label_parts.append(f"{arc.delay_distribution}")

        label = "\\n".join(label_parts)
        dot.append(f'  "{arc.source}" -> "{arc.target}" [label="{label}"];')

    dot.append("}")
    dot_str = "\n".join(dot)

    if filepath:
        with open(filepath, 'w') as f:
            f.write(dot_str)

    return dot_str

get_arc(source, target)

Get a specific arc by source and target name.

Source code in src\ergminer\erg_structure.py

def get_arc(self, source, target):
    """Get a specific arc by source and target name."""
    for arc in self.arcs:
        if arc.source == source and arc.target == target:
            return arc
    return None

get_arcs_from(source)

Get all arcs originating from a source node.

Source code in src\ergminer\erg_structure.py

def get_arcs_from(self, source):
    """Get all arcs originating from a source node."""
    return list(self._arcs_from[source])

get_arcs_to(target)

Get all arcs leading to a target node.

Source code in src\ergminer\erg_structure.py

def get_arcs_to(self, target):
    """Get all arcs leading to a target node."""
    return list(self._arcs_to[target])

get_node(name)

Get a node by name.

Source code in src\ergminer\erg_structure.py

def get_node(self, name):
    """Get a node by name."""
    return self.nodes.get(name)

get_statistics()

Get statistics about the ERG.

Source code in src\ergminer\erg_structure.py

def get_statistics(self):
    """Get statistics about the ERG."""
    return {
        'num_nodes': len(self.nodes),
        'num_arcs': len(self.arcs),
        'num_state_variables': len(self.state_variables),
        'num_start_events': len(self.start_events),
        'num_end_events': len(self.end_events),
        'immediate_arcs': sum(1 for arc in self.arcs if arc.is_immediate),
        'delayed_arcs': sum(1 for arc in self.arcs if not arc.is_immediate),
        'guarded_arcs': sum(1 for arc in self.arcs if arc.guard_condition is not None)
    }

print_summary(verbose=True)

Print a human-readable summary of the ERG.

Source code in src\ergminer\erg_structure.py

def print_summary(self, verbose=True):
    """Print a human-readable summary of the ERG."""
    if not verbose:
        return
    stats = self.get_statistics()

    print(f"\n{'='*60}")
    print(f"Event Relationship Graph: {self.name}")
    print(f"{'='*60}")

    print(f"\nNodes (Events): {stats['num_nodes']}")
    for node in self.nodes.values():
        start_marker = " [START]" if node.name in self.start_events else ""
        end_marker = " [END]" if node.name in self.end_events else ""
        print(f"  - {node.name} ({node.event_type}){start_marker}{end_marker}")
        if node.name not in self.start_events and node.name not in self.end_events:
            if node.state_update_equations:
                eqs = "; ".join(node.state_update_equations)
                print(f"      {{ {eqs} }}")
            else:
                print("      { }")

    print(f"\nState Variables: {stats['num_state_variables']}")
    for sv in self.state_variables.values():
        print(f"  - {sv.name}: {sv.variable_type} for {sv.resource} (init={sv.initial_value})")

    print(f"\nArcs (Transitions): {stats['num_arcs']}")
    print(f"  - Immediate: {stats['immediate_arcs']}")
    print(f"  - Delayed: {stats['delayed_arcs']}")
    print(f"  - Guarded: {stats['guarded_arcs']}")

    print("\nArc Details:")
    for arc in self.arcs:
        guard_str = f" [Guard: {arc.guard_condition}]" if arc.guard_condition else ""

        if arc.is_immediate:
            delay_str = " [Immediate]"
        else:
            delay_str = f" [{arc.delay_distribution}, mean={arc.mean_delay:.3f}]"

        updates_str = ""
        if arc.state_updates:
            updates_str = f" [Updates: {', '.join(arc.state_updates)}]"

        print(f"  {arc.source} -> {arc.target} (p={arc.probability:.3f}){guard_str}{delay_str}{updates_str}")

    print(f"\n{'='*60}\n")

set_end_event(event_name)

Mark an event as an end event.

Source code in src\ergminer\erg_structure.py

def set_end_event(self, event_name):
    """Mark an event as an end event."""
    self.end_events.add(event_name)

set_start_event(event_name)

Mark an event as a start event.

Source code in src\ergminer\erg_structure.py

def set_start_event(self, event_name):
    """Mark an event as a start event."""
    self.start_events.add(event_name)

to_dict()

Convert ERG to dictionary.

Source code in src\ergminer\erg_structure.py

def to_dict(self):
    """Convert ERG to dictionary."""
    return {
        'name': self.name,
        'nodes': [node.to_dict() for node in self.nodes.values()],
        'arcs': [arc.to_dict() for arc in self.arcs],
        'state_variables': [sv.to_dict() for sv in self.state_variables.values()],
        'start_events': list(self.start_events),
        'end_events': list(self.end_events),
        'statistics': self.get_statistics()
    }

to_json(filepath=None)

Export ERG to JSON format.

Parameters:

Name	Type	Description	Default
filepath		Optional path to save JSON file	None

Returns:

Type	Description
	JSON string representation

Source code in src\ergminer\erg_structure.py

def to_json(self, filepath=None):
    """
    Export ERG to JSON format.

    Args:
        filepath: Optional path to save JSON file

    Returns:
        JSON string representation
    """
    erg_dict = self.to_dict()
    json_str = json.dumps(erg_dict, indent=2)

    if filepath:
        with open(filepath, 'w') as f:
            f.write(json_str)

    return json_str

ERGNode

ergminer.ERGNode

Represents a node (event) in the ERG.

Source code in src\ergminer\erg_structure.py

class ERGNode:
    """Represents a node (event) in the ERG."""

    def __init__(self, name, event_type, original_activity, resource=None,
                 frequency=0, state_update_equations=None):
        self.name = name
        self.event_type = event_type  # e.g. 'Arrive', 'Run', 'End Process'
        self.original_activity = original_activity
        self.resource = resource
        self.frequency = frequency
        self.state_update_equations = state_update_equations if state_update_equations is not None else [] # List of state update equation strings displayed inside this node

    def to_dict(self):
        """Convert node to dictionary."""
        return {
            'name': self.name,
            'event_type': self.event_type,
            'original_activity': self.original_activity,
            'resource': self.resource,
            'frequency': self.frequency,
            'state_update_equations': self.state_update_equations
        }

to_dict()

Convert node to dictionary.

Source code in src\ergminer\erg_structure.py

def to_dict(self):
    """Convert node to dictionary."""
    return {
        'name': self.name,
        'event_type': self.event_type,
        'original_activity': self.original_activity,
        'resource': self.resource,
        'frequency': self.frequency,
        'state_update_equations': self.state_update_equations
    }

ERGArc

ergminer.ERGArc

Represents an arc (transition) in the ERG.

Source code in src\ergminer\erg_structure.py

class ERGArc:
    """Represents an arc (transition) in the ERG."""

    def __init__(self, source, target, probability=1.0, guard_condition=None,
                 is_immediate=True, delay_distribution=None, distribution_params=None,
                 mean_delay=None, state_updates=None, arc_type='DF'):
        self.source = source
        self.target = target
        self.probability = probability
        self.guard_condition = guard_condition
        self.is_immediate = is_immediate
        self.delay_distribution = delay_distribution
        self.distribution_params = distribution_params
        self.mean_delay = mean_delay
        self.state_updates = state_updates if state_updates is not None else []
        self.arc_type = arc_type  # 'DF' = within-case, 'DF_Res' = cross-case resource

        # Validate: every arc must be either immediate OR delayed, never both.
        # Immediate arcs have zero delay (e.g. routing decisions).
        # Delayed arcs carry a fitted probability distribution (e.g. service times).
        if self.is_immediate and self.delay_distribution not in (None, 'immediate'):
            raise ValueError(
                f"Arc {self.source} -> {self.target}: "
                f"is_immediate=True but delay_distribution='{self.delay_distribution}'. "
                f"Immediate arcs must have delay_distribution=None or 'immediate'."
            )

        # Standardise fields after validation
        if self.is_immediate:
            self.delay_distribution = 'immediate'
            self.distribution_params = None
            self.mean_delay = 0.0
        elif self.mean_delay is None:
            self.mean_delay = 0.0

    def to_dict(self):
        """Convert arc to dictionary."""
        return {
            'source': self.source,
            'target': self.target,
            'probability': self.probability,
            'guard_condition': self.guard_condition,
            'is_immediate': self.is_immediate,
            'delay_distribution': self.delay_distribution,
            'distribution_params': self.distribution_params,
            'mean_delay': self.mean_delay,
            'state_updates': self.state_updates,
            'arc_type': self.arc_type
        }

to_dict()

Convert arc to dictionary.

Source code in src\ergminer\erg_structure.py

def to_dict(self):
    """Convert arc to dictionary."""
    return {
        'source': self.source,
        'target': self.target,
        'probability': self.probability,
        'guard_condition': self.guard_condition,
        'is_immediate': self.is_immediate,
        'delay_distribution': self.delay_distribution,
        'distribution_params': self.distribution_params,
        'mean_delay': self.mean_delay,
        'state_updates': self.state_updates,
        'arc_type': self.arc_type
    }

ERGStateVariable

ergminer.ERGStateVariable

Represents a state variable in the ERG.

Source code in src\ergminer\erg_structure.py

class ERGStateVariable:
    """Represents a state variable in the ERG."""

    def __init__(self, name, variable_type, resource, initial_value):
        self.name = name
        self.variable_type = variable_type  # 'Server_Busy' or 'Queue_Length'
        self.resource = resource
        self.initial_value = initial_value

    def to_dict(self):
        """Convert state variable to dictionary."""
        return {
            'name': self.name,
            'variable_type': self.variable_type,
            'resource': self.resource,
            'initial_value': self.initial_value
        }

to_dict()

Convert state variable to dictionary.

Source code in src\ergminer\erg_structure.py

def to_dict(self):
    """Convert state variable to dictionary."""
    return {
        'name': self.name,
        'variable_type': self.variable_type,
        'resource': self.resource,
        'initial_value': self.initial_value
    }

ConformanceResult

ergminer.ConformanceResult

Holds all conformance scores and supporting detail tables.

Attributes

erg_name : str n_simulations : int checker_results : dict — raw results dict from ERGConformanceChecker summary_df : pd.DataFrame — one row per check (Check, Score, Pass, Detail) overall_score : float — unweighted mean of all check scores

Source code in src\ergminer\conformance_testing.py

class ConformanceResult:
    """
    Holds all conformance scores and supporting detail tables.

    Attributes
    ----------
    erg_name        : str
    n_simulations   : int
    checker_results : dict             — raw results dict from ERGConformanceChecker
    summary_df      : pd.DataFrame     — one row per check (Check, Score, Pass, Detail)
    overall_score   : float            — unweighted mean of all check scores
    """

    def __init__(self):
        self.erg_name        = ''
        self.n_simulations   = 0
        self.checker_results = {}
        self.summary_df      = None
        self.overall_score   = None
        self.group_scores    = {}   # {group_name: {'score': float, 'pass': bool, 'checks': int}}

    @staticmethod
    def _safe_score(numerator, denominator, invert=True):
        """Return 1 - num/denom (default) or num/denom, guarding division by zero."""
        if denominator == 0:
            return 1.0
        ratio = numerator / denominator
        return round(1.0 - ratio if invert else ratio, 6)

    # ------------------------------------------------------------------
    # Reporting
    # ------------------------------------------------------------------
    def print_report(self):
        """Print a human-readable conformance report to stdout."""
        W = 80
        print()
        print('=' * W)
        print(f"  ERG CONFORMANCE REPORT  —  {self.erg_name}")
        print('=' * W)
        print(f"  Simulations run : {self.n_simulations}")
        print()
        if self.summary_df is not None and not self.summary_df.empty:
            # Print checks grouped
            check_to_group = {
                ck: grp
                for grp, checks in CHECK_GROUPS.items()
                for ck in checks
            }
            printed_groups = set()
            for _, row in self.summary_df.iterrows():
                raw_key = row.get('key', '')
                grp = check_to_group.get(raw_key, 'Other')
                if grp not in printed_groups:
                    grp_info = self.group_scores.get(grp, {})
                    grp_score = grp_info.get('score')
                    grp_pass  = grp_info.get('pass')
                    grp_str   = f"  [{grp_score:.4f}]" if grp_score is not None else ''
                    grp_tick  = '\u2713' if grp_pass else '\u2717'
                    print(f"  {grp_tick} {grp}{grp_str}")
                    printed_groups.add(grp)
                score_str = f"{row['Score']:.4f}" if isinstance(row['Score'], float) else str(row['Score'])
                print(f"    {row['Pass']}  [{score_str}]  {row['Check']}")
                print(f"               {row['Detail']}")
            print('=' * W)
            passed = sum(1 for _, r in self.summary_df.iterrows() if '\u2713' in r['Pass'])
            total  = len(self.summary_df)
            overall_str = f"  |  Mean score: {self.overall_score:.4f}" if self.overall_score is not None else ''
            print(f"  Overall: {passed}/{total} checks passed{overall_str}")
        print('=' * W)
        print()

    # ------------------------------------------------------------------
    # Save detail CSVs
    # ------------------------------------------------------------------

    def save_detail_csvs(self, output_dir, log_stem):
        """
        Write one CSV per conformance check into *output_dir*.
        Files are named  <log_stem>_conf_<N>_<check>.csv

        Parameters
        ----------
        output_dir : str | Path
        log_stem   : str  — prefix derived from the ERG / log filename
        """
        from pathlib import Path
        out = Path(output_dir)
        out.mkdir(parents=True, exist_ok=True)
        cr = self.checker_results

        def _save(name, rows):
            pd.DataFrame(rows).to_csv(out / f'{log_stem}_{name}.csv', index=False)

        # Summary — inject group header rows so checks appear under their group.
        # Rows are sorted by group order (as defined in CHECK_GROUPS) so all
        # checks belonging to a group appear contiguously under its header.
        if self.summary_df is not None:
            check_to_group = {
                ck: grp
                for grp, checks in CHECK_GROUPS.items()
                for ck in checks
            }
            group_order = {grp: i for i, grp in enumerate(CHECK_GROUPS)}

            # Sort the check rows by group order, preserving within-group run order
            check_rows = list(self.summary_df.iterrows())
            check_rows.sort(key=lambda r: group_order.get(
                check_to_group.get(r[1].get('key', ''), 'Other'), 999))

            summary_rows = []
            seen_groups = set()
            for _, row in check_rows:
                grp = check_to_group.get(row.get('key', ''), 'Other')
                if grp not in seen_groups:
                    grp_info = self.group_scores.get(grp, {})
                    summary_rows.append({
                        'key':    f'GROUP:{grp}',
                        'Group':  grp,
                        'Check':  f'--- {grp} ---',
                        'Score':  round(grp_info['score'], 4) if grp_info.get('score') is not None else '',
                        'Pass':   ('\u2713 PASS' if grp_info.get('pass') else '\u2717 FAIL')
                                  if grp_info else '',
                        'Detail': f"Group score (mean of {grp_info.get('checks', 0)} checks)",
                    })
                    seen_groups.add(grp)
                summary_rows.append(dict(row))
            pd.DataFrame(summary_rows).to_csv(
                out / f'{log_stem}_conformance_summary.csv', index=False
            )

        # Group summary
        if self.group_scores:
            group_rows = [
                {'group': grp, 'score': round(info['score'], 4),
                 'pass': info['pass'], 'n_checks': info['checks']}
                for grp, info in self.group_scores.items()
            ]
            pd.DataFrame(group_rows).to_csv(
                out / f'{log_stem}_conformance_group_summary.csv', index=False
            )

        # 1. Trace-to-node mapping
        r = cr.get('trace_to_node_mapping', {})
        missing = r.get('missing_activities', [])
        if missing:
            rows1 = [{'result': 'FAIL', 'score': r.get('score'), 'detail': r.get('detail', ''),
                      'missing_activity': a} for a in missing]
        else:
            rows1 = [{'result': 'PASS', 'score': r.get('score'), 'detail': r.get('detail', ''),
                      'missing_activity': None}]
        _save('conf_1_trace_node_mapping', rows1)

        # 2. Delay arc timing — per-arc KS results
        r = cr.get('delay_arc_timing', {})
        _save('conf_2_delay_arc_timing', list(r.get('arc_results', {}).values()))

        # 3. Arc type coverage — per-arc pass/fail
        r = cr.get('arc_type_coverage', {})
        _save('conf_3_arc_type_coverage',
              list(r.get('arc_detail', {}).values()))

        # 4. Prefix closure — one row per unique prefix with valid_in_erg status
        r = cr.get('prefix_closure', {})
        _save('conf_4_prefix_closure', r.get('all_prefix_detail', []))

        # 5. Variant coverage — one row per variant
        r = cr.get('variant_coverage', {})
        _save('conf_5_variant_coverage', r.get('variant_detail', []))

        # 6. Node visit frequency — per-activity orig vs sim counts
        r = cr.get('node_visit_frequency', {})
        _save('conf_6_node_visit_frequency',
              [{'activity': a, **counts}
               for a, counts in r.get('activity_counts', {}).items()])

        # 7. Trace length distribution — scalar summary
        r = cr.get('trace_length_distribution', {})
        _save('conf_7_trace_length_distribution',
              [{k: v for k, v in r.items() if k != 'detail'}])

        # 8. Arc firing frequency — per-arc orig vs sim frequency
        r = cr.get('arc_firing_frequency', {})
        _save('conf_8_arc_firing_frequency',
              list(r.get('edge_detail', {}).values()))

        # 9. Alignment fitness — scalar summary
        r = cr.get('alignment_fitness', {})
        _save('conf_9_alignment_fitness',
              [{k: v for k, v in r.items() if k != 'detail'}])

        # 10. Escaping edges — never-fired arcs with arcType and reason
        r = cr.get('escaping_edges', {})
        _save('conf_10_escaping_edges', r.get('never_fired_edges', []))

        # 11. Routing probability fidelity — per-branch comparison
        r = cr.get('routing_probability_fidelity', {})
        arc_rows = r.get('arc_detail', [])
        _save('conf_11_routing_probability_fidelity',
              arc_rows if arc_rows
              else [{'note': 'No probabilistic branching nodes found in ERG — routing probability fidelity not applicable'}])

print_report()

Print a human-readable conformance report to stdout.

Source code in src\ergminer\conformance_testing.py

def print_report(self):
    """Print a human-readable conformance report to stdout."""
    W = 80
    print()
    print('=' * W)
    print(f"  ERG CONFORMANCE REPORT  —  {self.erg_name}")
    print('=' * W)
    print(f"  Simulations run : {self.n_simulations}")
    print()
    if self.summary_df is not None and not self.summary_df.empty:
        # Print checks grouped
        check_to_group = {
            ck: grp
            for grp, checks in CHECK_GROUPS.items()
            for ck in checks
        }
        printed_groups = set()
        for _, row in self.summary_df.iterrows():
            raw_key = row.get('key', '')
            grp = check_to_group.get(raw_key, 'Other')
            if grp not in printed_groups:
                grp_info = self.group_scores.get(grp, {})
                grp_score = grp_info.get('score')
                grp_pass  = grp_info.get('pass')
                grp_str   = f"  [{grp_score:.4f}]" if grp_score is not None else ''
                grp_tick  = '\u2713' if grp_pass else '\u2717'
                print(f"  {grp_tick} {grp}{grp_str}")
                printed_groups.add(grp)
            score_str = f"{row['Score']:.4f}" if isinstance(row['Score'], float) else str(row['Score'])
            print(f"    {row['Pass']}  [{score_str}]  {row['Check']}")
            print(f"               {row['Detail']}")
        print('=' * W)
        passed = sum(1 for _, r in self.summary_df.iterrows() if '\u2713' in r['Pass'])
        total  = len(self.summary_df)
        overall_str = f"  |  Mean score: {self.overall_score:.4f}" if self.overall_score is not None else ''
        print(f"  Overall: {passed}/{total} checks passed{overall_str}")
    print('=' * W)
    print()

save_detail_csvs(output_dir, log_stem)

Write one CSV per conformance check into output_dir. Files are named conf_.csv

Parameters

output_dir : str | Path log_stem : str — prefix derived from the ERG / log filename

Source code in src\ergminer\conformance_testing.py

def save_detail_csvs(self, output_dir, log_stem):
    """
    Write one CSV per conformance check into *output_dir*.
    Files are named  <log_stem>_conf_<N>_<check>.csv

    Parameters
    ----------
    output_dir : str | Path
    log_stem   : str  — prefix derived from the ERG / log filename
    """
    from pathlib import Path
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    cr = self.checker_results

    def _save(name, rows):
        pd.DataFrame(rows).to_csv(out / f'{log_stem}_{name}.csv', index=False)

    # Summary — inject group header rows so checks appear under their group.
    # Rows are sorted by group order (as defined in CHECK_GROUPS) so all
    # checks belonging to a group appear contiguously under its header.
    if self.summary_df is not None:
        check_to_group = {
            ck: grp
            for grp, checks in CHECK_GROUPS.items()
            for ck in checks
        }
        group_order = {grp: i for i, grp in enumerate(CHECK_GROUPS)}

        # Sort the check rows by group order, preserving within-group run order
        check_rows = list(self.summary_df.iterrows())
        check_rows.sort(key=lambda r: group_order.get(
            check_to_group.get(r[1].get('key', ''), 'Other'), 999))

        summary_rows = []
        seen_groups = set()
        for _, row in check_rows:
            grp = check_to_group.get(row.get('key', ''), 'Other')
            if grp not in seen_groups:
                grp_info = self.group_scores.get(grp, {})
                summary_rows.append({
                    'key':    f'GROUP:{grp}',
                    'Group':  grp,
                    'Check':  f'--- {grp} ---',
                    'Score':  round(grp_info['score'], 4) if grp_info.get('score') is not None else '',
                    'Pass':   ('\u2713 PASS' if grp_info.get('pass') else '\u2717 FAIL')
                              if grp_info else '',
                    'Detail': f"Group score (mean of {grp_info.get('checks', 0)} checks)",
                })
                seen_groups.add(grp)
            summary_rows.append(dict(row))
        pd.DataFrame(summary_rows).to_csv(
            out / f'{log_stem}_conformance_summary.csv', index=False
        )

    # Group summary
    if self.group_scores:
        group_rows = [
            {'group': grp, 'score': round(info['score'], 4),
             'pass': info['pass'], 'n_checks': info['checks']}
            for grp, info in self.group_scores.items()
        ]
        pd.DataFrame(group_rows).to_csv(
            out / f'{log_stem}_conformance_group_summary.csv', index=False
        )

    # 1. Trace-to-node mapping
    r = cr.get('trace_to_node_mapping', {})
    missing = r.get('missing_activities', [])
    if missing:
        rows1 = [{'result': 'FAIL', 'score': r.get('score'), 'detail': r.get('detail', ''),
                  'missing_activity': a} for a in missing]
    else:
        rows1 = [{'result': 'PASS', 'score': r.get('score'), 'detail': r.get('detail', ''),
                  'missing_activity': None}]
    _save('conf_1_trace_node_mapping', rows1)

    # 2. Delay arc timing — per-arc KS results
    r = cr.get('delay_arc_timing', {})
    _save('conf_2_delay_arc_timing', list(r.get('arc_results', {}).values()))

    # 3. Arc type coverage — per-arc pass/fail
    r = cr.get('arc_type_coverage', {})
    _save('conf_3_arc_type_coverage',
          list(r.get('arc_detail', {}).values()))

    # 4. Prefix closure — one row per unique prefix with valid_in_erg status
    r = cr.get('prefix_closure', {})
    _save('conf_4_prefix_closure', r.get('all_prefix_detail', []))

    # 5. Variant coverage — one row per variant
    r = cr.get('variant_coverage', {})
    _save('conf_5_variant_coverage', r.get('variant_detail', []))

    # 6. Node visit frequency — per-activity orig vs sim counts
    r = cr.get('node_visit_frequency', {})
    _save('conf_6_node_visit_frequency',
          [{'activity': a, **counts}
           for a, counts in r.get('activity_counts', {}).items()])

    # 7. Trace length distribution — scalar summary
    r = cr.get('trace_length_distribution', {})
    _save('conf_7_trace_length_distribution',
          [{k: v for k, v in r.items() if k != 'detail'}])

    # 8. Arc firing frequency — per-arc orig vs sim frequency
    r = cr.get('arc_firing_frequency', {})
    _save('conf_8_arc_firing_frequency',
          list(r.get('edge_detail', {}).values()))

    # 9. Alignment fitness — scalar summary
    r = cr.get('alignment_fitness', {})
    _save('conf_9_alignment_fitness',
          [{k: v for k, v in r.items() if k != 'detail'}])

    # 10. Escaping edges — never-fired arcs with arcType and reason
    r = cr.get('escaping_edges', {})
    _save('conf_10_escaping_edges', r.get('never_fired_edges', []))

    # 11. Routing probability fidelity — per-branch comparison
    r = cr.get('routing_probability_fidelity', {})
    arc_rows = r.get('arc_detail', [])
    _save('conf_11_routing_probability_fidelity',
          arc_rows if arc_rows
          else [{'note': 'No probabilistic branching nodes found in ERG — routing probability fidelity not applicable'}])