Skip to content

api_reference.md

Latest commit

 

History

History
869 lines (692 loc) · 24 KB

api_reference.md

File metadata and controls

869 lines (692 loc) · 24 KB
Raw

Heraclitus API Reference

This document provides a comprehensive reference for the Heraclitus API, including all modules, classes, and functions.

Table of Contents

Data Module

EventLog

The EventLog class is the central data structure in Heraclitus, representing event log data with standardized access patterns.

Constructor

EventLog(
    data: Union[pd.DataFrame, str],
    case_id_column: str = "case_id",
    activity_column: str = "activity",
    timestamp_column: str = "timestamp",
    duplicate_handling: Literal["error", "warn", "keep_first", "keep_last"] = "warn",
    **kwargs: Any
) -> None

  • Parameters:

    • data: Data source (pandas DataFrame or file path)
    • case_id_column: Column name for case identifiers
    • activity_column: Column name for activities
    • timestamp_column: Column name for timestamps
    • duplicate_handling: How to handle duplicate events
    • **kwargs: Additional parameters passed to data loading functions

  • Raises:

    • ValueError: If required columns are missing or data format is invalid
    • TypeError: If the data type is not supported

Class Methods

@classmethod
from_dataframe(cls, df: pd.DataFrame, **kwargs: Any) -> EventLog

  • Parameters:

    • df: Source DataFrame containing event data
    • **kwargs: Additional parameters passed to the EventLog constructor

  • Returns: A new EventLog instance

@classmethod
from_csv(cls, filepath: str, **kwargs: Any) -> EventLog

  • Parameters:

    • filepath: Path to the CSV file
    • **kwargs: Additional parameters passed to pd.read_csv and the EventLog constructor

  • Returns: A new EventLog instance

@classmethod
from_xes(cls, file_path: str, **kwargs: Any) -> EventLog

  • Parameters:

    • file_path: Path to the XES file
    • **kwargs: Additional parameters passed to the import_xes function

  • Returns: A new EventLog instance

  • Raises: FileNotFoundError if the XES file is not found, ValueError if the file is not valid XES

Instance Methods

to_dataframe() -> pd.DataFrame
  • Returns: A copy of the internal DataFrame
filter_cases(case_ids: List[str]) -> EventLog

  • Parameters:

    • case_ids: List of case IDs to include

  • Returns: A new EventLog instance with only the specified cases

filter_activities(activities: List[str]) -> EventLog

  • Parameters:

    • activities: List of activities to include

  • Returns: A new EventLog instance with only the specified activities

filter_time_range(
    start_time: Optional[pd.Timestamp] = None,
    end_time: Optional[pd.Timestamp] = None
) -> EventLog

  • Parameters:

    • start_time: Start of the time range (inclusive)
    • end_time: End of the time range (inclusive)

  • Returns: A new EventLog instance with only events in the specified time range

get_attributes() -> List[str]
  • Returns: List of attribute column names
add_attribute(column_name: str, data: List[Any]) -> None

  • Parameters:

    • column_name: Name of the new attribute column
    • data: Data for the new attribute column

  • Raises:

    • ValueError: If the data length doesn't match the number of rows, or if the column already exists
to_xes(file_path: str, pretty_print: bool = True) -> None

  • Parameters:

    • file_path: Path to save the XES file
    • pretty_print: Whether to format the XML with indentation

  • Raises:

    • ValueError: If the EventLog is invalid or missing required columns
case_count() -> int
  • Returns: The number of unique cases in the log
activity_count() -> int
  • Returns: The number of unique activities in the log
get_unique_activities() -> List[str]
  • Returns: A list of unique activities in the log
get_unique_cases() -> List[str]
  • Returns: A list of unique case IDs in the log

DuckDBConnector

The DuckDBConnector class provides utilities for handling large event logs using DuckDB.

Constructor

DuckDBConnector(db_path: Optional[Union[str, Path]] = None)
  • Parameters:
    • db_path: Optional path to the database file. If None, an in-memory database is used.

Methods

load_csv(
    filepath: Union[str, Path],
    table_name: str,
    case_id_column: str = "case_id",
    activity_column: str = "activity",
    timestamp_column: str = "timestamp",
    timestamp_format: Optional[str] = None,
    delimiter: str = ",",
    header: bool = True,
    **kwargs: Any
) -> None
  • Parameters:
    • filepath: Path to the CSV file
    • table_name: Name for the table in DuckDB
    • case_id_column: Name of the case ID column
    • activity_column: Name of the activity column
    • timestamp_column: Name of the timestamp column
    • timestamp_format: Optional format string for parsing timestamps
    • delimiter: CSV delimiter character
    • header: Whether the CSV file has a header row
    • **kwargs: Additional parameters passed to duckdb.read_csv
load_dataframe(
    df: pd.DataFrame,
    table_name: str,
    case_id_column: str = "case_id",
    activity_column: str = "activity",
    timestamp_column: str = "timestamp"
) -> None
  • Parameters:
    • df: Source DataFrame
    • table_name: Name for the table in DuckDB
    • case_id_column: Name of the case ID column
    • activity_column: Name of the activity column
    • timestamp_column: Name of the timestamp column
query_to_eventlog(
    query: str,
    table_name: Optional[str] = None,
    case_id_column: Optional[str] = None,
    activity_column: Optional[str] = None,
    timestamp_column: Optional[str] = None
) -> EventLog

  • Parameters:

    • query: SQL query to execute
    • table_name: Optional table name to get metadata from
    • case_id_column: Optional case ID column name (overrides table metadata)
    • activity_column: Optional activity column name (overrides table metadata)
    • timestamp_column: Optional timestamp column name (overrides table metadata)

  • Returns: An EventLog instance containing the query results

  • Raises: ValueError if required column information is not provided

get_eventlog(
    table_name: str,
    case_ids: Optional[List[str]] = None,
    activities: Optional[List[str]] = None,
    start_time: Optional[pd.Timestamp] = None,
    end_time: Optional[pd.Timestamp] = None,
    limit: Optional[int] = None
) -> EventLog

  • Parameters:

    • table_name: Name of the table
    • case_ids: Optional list of case IDs to filter by
    • activities: Optional list of activities to filter by
    • start_time: Optional start timestamp for filtering
    • end_time: Optional end timestamp for filtering
    • limit: Optional row limit for the query

  • Returns: An EventLog instance containing the filtered data

  • Raises: ValueError if the table is not registered

get_unique_values(
    table_name: str,
    column_name: str,
    limit: Optional[int] = None
) -> List[Any]

  • Parameters:

    • table_name: Name of the table
    • column_name: Name of the column
    • limit: Optional limit for the number of unique values

  • Returns: A list of unique values

  • Raises: ValueError if the table is not registered

get_case_count(table_name: str) -> int

  • Parameters:

    • table_name: Name of the table

  • Returns: The number of unique cases

get_activity_count(table_name: str) -> int

  • Parameters:

    • table_name: Name of the table

  • Returns: The number of unique activities

get_time_range(table_name: str) -> Tuple[pd.Timestamp, pd.Timestamp]

  • Parameters:

    • table_name: Name of the table

  • Returns: A tuple of (min_timestamp, max_timestamp)

get_table_info(table_name: str) -> Dict[str, Any]

  • Parameters:

    • table_name: Name of the table

  • Returns: A dictionary containing table information

execute(query: str) -> duckdb.DuckDBPyRelation

  • Parameters:

    • query: SQL query to execute

  • Returns: DuckDB query result

close() -> None

Closes the DuckDB connection.

Helper Functions

eventlog_to_duckdb(
    event_log: EventLog,
    connector: DuckDBConnector,
    table_name: str
) -> None
  • Parameters:
    • event_log: The EventLog to save
    • connector: DuckDBConnector instance
    • table_name: Name for the table in DuckDB

XES Format Support

Functions for importing and exporting XES (eXtensible Event Stream) format files, which is the IEEE 1849-2016 standard for process mining event logs.

import_xes(file_path: str, timestamp_format: Optional[str] = None) -> EventLog

  • Parameters:

    • file_path: Path to the XES file
    • timestamp_format: Optional format string for parsing timestamps

  • Returns: EventLog object created from the XES file

  • Raises:

    • FileNotFoundError: If the XES file is not found
    • ValueError: If the file is not a valid XES file
export_xes(event_log: EventLog, file_path: str, pretty_print: bool = True) -> None

  • Parameters:

    • event_log: The EventLog to export
    • file_path: Path to save the XES file
    • pretty_print: Whether to format the XML with indentation

  • Raises:

    • ValueError: If the EventLog is invalid or missing required columns

XES Attribute Class

class XESAttribute:
    """Class representing an XES attribute."""
    
    def __init__(self, key: str, value: Any, type_name: str = "string"):
        """
        Initialize an XES attribute.
        
        Args:
            key: Attribute key
            value: Attribute value
            type_name: Type of the attribute (string, date, int, float, boolean)
        """
        
    def to_xml(self) -> ET.Element:
        """
        Convert the attribute to an XML element.
        
        Returns:
            ElementTree Element representing the attribute
        """

Metrics Module

Time Metrics

calculate_cycle_time(
    event_log: EventLog,
    case_id: Optional[str] = None,
    start_activity: Optional[str] = None,
    end_activity: Optional[str] = None,
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds",
    include_stats: bool = False
) -> Union[float, Dict[str, float]]

  • Parameters:

    • event_log: The event log to analyze
    • case_id: Optional specific case ID to analyze
    • start_activity: Optional specific start activity (first occurrence)
    • end_activity: Optional specific end activity (last occurrence)
    • unit: Time unit for the result
    • include_stats: Whether to include additional statistics

  • Returns:

    • If include_stats is False, returns the mean cycle time.
    • If include_stats is True, returns a dictionary with mean, median, min, max, and standard deviation of cycle times.

  • Raises:

    • ValueError: If case_id is provided but not found, or if activities are not found
calculate_waiting_time(
    event_log: EventLog,
    activity: str,
    case_id: Optional[str] = None,
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds",
    include_stats: bool = False
) -> Union[float, Dict[str, float]]

  • Parameters:

    • event_log: The event log to analyze
    • activity: The activity to calculate waiting time for
    • case_id: Optional specific case ID to analyze
    • unit: Time unit for the result
    • include_stats: Whether to include additional statistics

  • Returns:

    • If include_stats is False, returns the mean waiting time.
    • If include_stats is True, returns a dictionary with mean, median, min, max, and standard deviation of waiting times.

  • Raises:

    • ValueError: If activity is not found, or if case_id is provided but not found
calculate_processing_time(
    event_log: EventLog,
    activity: str,
    case_id: Optional[str] = None,
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds",
    include_stats: bool = False,
    skip_last_events: bool = True
) -> Union[float, Dict[str, float]]

  • Parameters:

    • event_log: The event log to analyze
    • activity: The activity to calculate processing time for
    • case_id: Optional specific case ID to analyze
    • unit: Time unit for the result
    • include_stats: Whether to include additional statistics
    • skip_last_events: Whether to skip activity occurrences that are the last in a case

  • Returns:

    • If include_stats is False, returns the mean processing time.
    • If include_stats is True, returns a dictionary with mean, median, min, max, and standard deviation of processing times.

  • Raises:

    • ValueError: If activity is not found, or if case_id is provided but not found

Statistics Module

Comparison Functions

compare_cycle_times(
    event_log: EventLog,
    group_by_attribute: str,
    start_activity: Optional[str] = None,
    end_activity: Optional[str] = None,
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds",
    test_type: Literal["parametric", "non_parametric"] = "parametric",
    alpha: float = 0.05
) -> Dict[str, Any]

  • Parameters:

    • event_log: The event log to analyze
    • group_by_attribute: Attribute to group cases by
    • start_activity: Optional specific start activity
    • end_activity: Optional specific end activity
    • unit: Time unit for the result
    • test_type: Type of statistical test to use
    • alpha: Significance level

  • Returns: A dictionary containing test results, including:

    • groups: List of group names
    • sample_sizes: Sample size for each group
    • means: Mean cycle time for each group
    • medians: Median cycle time for each group
    • test_name: Name of the statistical test used
    • statistic: Test statistic
    • p_value: P-value of the test
    • significant: Whether the difference is significant
    • effect_size: Effect size measurement
    • effect_size_name: Name of the effect size metric
    • effect_interpretation: Interpretation of the effect size

  • Raises:

    • ValueError: If group_by_attribute doesn't exist or has fewer than 2 groups
bottleneck_analysis(
    event_log: EventLog,
    method: Literal["waiting_time", "processing_time", "frequency"] = "waiting_time",
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds",
    top_n: int = 3
) -> Dict[str, Any]

  • Parameters:

    • event_log: The event log to analyze
    • method: Method to identify bottlenecks
    • unit: Time unit for time-based methods
    • top_n: Number of top bottlenecks to identify

  • Returns: A dictionary containing bottleneck analysis results:

    • method: Method used for analysis
    • bottlenecks: List of bottleneck activities
    • metrics: Metrics for each activity

  • Raises:

    • ValueError: If an invalid method is specified or no activities are found
fit_distribution(
    event_log: EventLog,
    data_type: Literal["cycle_time", "waiting_time", "processing_time"] = "cycle_time",
    distribution: Literal["exponential", "normal", "lognormal", "gamma", "weibull"] = "exponential",
    activity: Optional[str] = None,
    unit: Literal["seconds", "minutes", "hours", "days"] = "seconds"
) -> Dict[str, Any]

  • Parameters:

    • event_log: The event log to analyze
    • data_type: Type of data to fit distribution to
    • distribution: Type of distribution to fit
    • activity: Required for waiting_time and processing_time
    • unit: Time unit for the analysis

  • Returns: A dictionary containing distribution fitting results:

    • distribution: Name of the distribution
    • params: Estimated distribution parameters
    • sse: Sum of squared errors (goodness of fit)
    • ks_statistic: Kolmogorov-Smirnov test statistic
    • ks_pvalue: Kolmogorov-Smirnov test p-value
    • data: Original data points
    • data_summary: Summary statistics of the data

  • Raises:

    • ValueError: If invalid parameters are provided

Visualization Module

Process Maps

visualize_process_map(
    event_log: EventLog,
    frequency_threshold: float = 0.0,
    title: Optional[str] = None,
    edge_color_metric: Optional[str] = None,
    performance_metric: Optional[str] = None,
    node_size_by: Optional[str] = None,
    figsize: Tuple[int, int] = (10, 8),
    **kwargs
) -> matplotlib.figure.Figure

  • Parameters:

    • event_log: The event log to visualize
    • frequency_threshold: Minimum frequency for edges to be shown (0.0-1.0)
    • title: Optional title for the process map
    • edge_color_metric: Metric for edge colors ("frequency" or "performance")
    • performance_metric: Performance metric if edge_color_metric is "performance"
    • node_size_by: How to determine node sizes ("frequency" or "duration")
    • figsize: Figure size in inches
    • **kwargs: Additional parameters passed to plotting functions

  • Returns: A matplotlib Figure object

plot_activity_frequency(
    event_log: EventLog,
    top_n: Optional[int] = None,
    title: Optional[str] = None,
    sort_by: Literal["frequency", "name"] = "frequency",
    color_palette: str = "blues",
    figsize: Tuple[int, int] = (10, 6),
    **kwargs
) -> matplotlib.figure.Figure

  • Parameters:

    • event_log: The event log to visualize
    • top_n: Number of top activities to show
    • title: Optional title for the chart
    • sort_by: How to sort activities
    • color_palette: Color palette to use
    • figsize: Figure size in inches
    • **kwargs: Additional parameters passed to plotting functions

  • Returns: A matplotlib Figure object

Interactive Visualizations

create_interactive_process_map(
    event_log: EventLog,
    frequency_threshold: float = 0.0,
    performance_metric: Optional[str] = None,
    time_unit: Literal["seconds", "minutes", "hours", "days"] = "hours",
    node_size_range: Tuple[int, int] = (30, 100),
    edge_width_range: Tuple[int, int] = (2, 10),
    layout: str = "force",
    **kwargs
) -> plotly.graph_objects.Figure

  • Parameters:

    • event_log: The event log to visualize
    • frequency_threshold: Minimum frequency for edges to be shown (0.0-1.0)
    • performance_metric: Optional performance metric for edge colors
    • time_unit: Time unit for performance metrics
    • node_size_range: Range of node sizes
    • edge_width_range: Range of edge widths
    • layout: Graph layout algorithm
    • **kwargs: Additional parameters for customization

  • Returns: A Plotly Figure object

create_cycle_time_distribution(
    event_log: EventLog,
    time_unit: Literal["seconds", "minutes", "hours", "days"] = "days",
    group_by: Optional[str] = None,
    title: Optional[str] = None,
    bins: int = 30,
    **kwargs
) -> plotly.graph_objects.Figure

  • Parameters:

    • event_log: The event log to visualize
    • time_unit: Time unit for cycle times
    • group_by: Optional attribute to group by
    • title: Optional title for the chart
    • bins: Number of histogram bins
    • **kwargs: Additional parameters for customization

  • Returns: A Plotly Figure object

create_activity_timeline(
    event_log: EventLog,
    top_n_cases: Optional[int] = None,
    sort_by: Literal["case_id", "duration", "start_time"] = "case_id",
    color_by: Literal["activity", "resource"] = "activity",
    **kwargs
) -> plotly.graph_objects.Figure

  • Parameters:

    • event_log: The event log to visualize
    • top_n_cases: Number of cases to show
    • sort_by: How to sort cases
    • color_by: Attribute to use for event colors
    • **kwargs: Additional parameters for customization

  • Returns: A Plotly Figure object

create_bottleneck_dashboard(
    event_log: EventLog,
    time_unit: Literal["seconds", "minutes", "hours", "days"] = "hours",
    include_waiting_time: bool = True,
    include_processing_time: bool = True,
    **kwargs
) -> plotly.graph_objects.Figure

  • Parameters:

    • event_log: The event log to analyze
    • time_unit: Time unit for metrics
    • include_waiting_time: Whether to include waiting time analysis
    • include_processing_time: Whether to include processing time analysis
    • **kwargs: Additional parameters for customization

  • Returns: A Plotly Figure object with multiple subplots

Machine Learning Module

Feature Engineering

class FeatureExtractor:
    """
    Extracts features from event logs for machine learning.
    """
    
    def __init__(
        self,
        event_log: EventLog,
        time_unit: Literal["seconds", "minutes", "hours", "days"] = "days"
    )
    
    def extract_case_features(
        self,
        include_duration: bool = True,
        include_activities: bool = True,
        include_resources: bool = False,
        include_attributes: Optional[List[str]] = None,
        include_timing: bool = True,
        one_hot_encode: bool = True
    ) -> pd.DataFrame
    
    def extract_sequence_features(
        self,
        window_size: int = 2,
        include_time_between: bool = True
    ) -> pd.DataFrame

Prediction

class OutcomePredictor:
    """
    Predicts process outcomes based on event log data.
    """
    
    def __init__(
        self,
        event_log: EventLog,
        outcome_attribute: str,
        model_type: str = "random_forest"
    )
    
    def train(
        self,
        features: Optional[List[str]] = None,
        test_size: float = 0.2,
        **kwargs
    ) -> Dict[str, Any]
    
    def predict(
        self,
        case_data: Union[pd.DataFrame, EventLog]
    ) -> np.ndarray
    
    def predict_proba(
        self,
        case_data: Union[pd.DataFrame, EventLog]
    ) -> np.ndarray
    
    def feature_importance(self) -> Dict[str, float]
class DurationPredictor:
    """
    Predicts case duration based on event log data.
    """
    
    def __init__(
        self,
        event_log: EventLog,
        model_type: str = "random_forest",
        time_unit: Literal["seconds", "minutes", "hours", "days"] = "days"
    )
    
    def train(
        self,
        features: Optional[List[str]] = None,
        test_size: float = 0.2,
        **kwargs
    ) -> Dict[str, Any]
    
    def predict(
        self,
        case_data: Union[pd.DataFrame, EventLog]
    ) -> np.ndarray
    
    def feature_importance(self) -> Dict[str, float]

Anomaly Detection

class ProcessAnomalyDetector:
    """
    Detects anomalies in process execution.
    """
    
    def __init__(
        self,
        event_log: EventLog,
        detection_method: Literal["isolation_forest", "lof", "one_class_svm"] = "isolation_forest"
    )
    
    def detect_anomalies(
        self,
        features: Optional[List[str]] = None,
        contamination: float = 0.05,
        **kwargs
    ) -> Dict[str, Any]
    
    def get_anomaly_scores(self) -> Dict[str, float]
    
    def explain_anomalies(
        self,
        top_n: int = 5
    ) -> Dict[str, Any]
def detect_variant_anomalies(
    event_log: EventLog,
    frequency_threshold: float = 0.01,
    include_statistics: bool = False
) -> Dict[str, Any]

  • Parameters:

    • event_log: The event log to analyze
    • frequency_threshold: Threshold for defining infrequent variants
    • include_statistics: Whether to include detailed statistics

  • Returns: A dictionary with detected variant anomalies:

    • anomalous_variants: List of anomalous variant sequences
    • normal_variants: List of normal variant sequences
    • anomalous_cases: List of case IDs with anomalous variants
    • statistics (if include_statistics is True): Additional statistics