diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f270208..8d769947 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +- Optimal alignments (`conformance::alignments`): + - `align_log` / `align_projection` / `align_trace` / `align_empty_trace` compute cost-optimal alignments via Dijkstra over a synchronous product net + - `compute_fitness` derives log/trace fitness for pre-computed alignments + - Configurable `AlignmentOptions` (`CostFunction`, `max_states`) + - Exposed as bindings +- Generic, reusable state-space search in `utils::dijkstra_search` (`SearchProblem` trait + `search`) +- `register_binding` macro accepts slice arguments (`&[T]` extracted as `Vec`) - New `ReadableOCEL` and `AppendableOCEL` traits; OCEL exporters (CSV, XML, SQL, JSON) and the JSON importer are generic over them - `SlimLinkedOCEL` implements `AppendableOCEL`, with auto-declare types, auto-grow attributes, and value coercion via new `OCELAttributeValue::try_coerce_to`; misordered streams are buffered and resolved on `finalize` - `import_ocel_json_into` and `import_ocel_xml_into` stream JSON / XML directly into any `AppendableOCEL` (e.g., `SlimLinkedOCEL`) without materializing an `OCEL` first; `SlimLinkedOCEL::import_from_*` uses the streaming paths automatically diff --git a/Cargo.lock b/Cargo.lock index cde4efbd..890ab8f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1540,6 +1540,11 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", +] [[package]] name = "hashlink" @@ -3117,7 +3122,7 @@ dependencies = [ "duckdb", "flate2", "graphviz-rust", - "hashbrown 0.15.5", + "hashbrown 0.17.1", "inventory", "itertools 0.14.0", "kuzu", @@ -3130,6 +3135,7 @@ dependencies = [ "rand 0.9.4", "rayon", "rusqlite", + "rustc-hash 2.1.2", "schemars 1.2.1", "serde", "serde_json", diff --git a/macros_process_mining/src/lib.rs b/macros_process_mining/src/lib.rs index 4b8d0403..d4e19064 100644 --- a/macros_process_mining/src/lib.rs +++ b/macros_process_mining/src/lib.rs @@ -228,7 +228,15 @@ pub fn register_binding(args: TokenStream, item: TokenStream) -> TokenStream { && !(BIG_TYPES_NAMES.iter().any(|tn| ty_as_str.ends_with(tn))); let type_without_ref = match &ty_no_life { Type::Reference(type_reference) if change_from_ref => { - *type_reference.elem.clone() + match &*type_reference.elem { + // `&[T]` -> extract a `Vec`; `&Vec` coerces back to `&[T]` at the + // call site. (Does not apply to `&[&str]`: `Vec<&str>` is not owned.) + Type::Slice(slice) => { + let elem = &slice.elem; + syn::parse_quote!(Vec<#elem>) + } + _ => *type_reference.elem.clone(), + } } x => x.clone(), }; diff --git a/process_mining/Cargo.toml b/process_mining/Cargo.toml index bd13e8b6..b14d14b9 100644 --- a/process_mining/Cargo.toml +++ b/process_mining/Cargo.toml @@ -19,7 +19,6 @@ chrono = { version = "0.4.40", features = ["serde"] } duckdb = { version = "1.2.1", optional = true, features = ["chrono"]} flate2 = "1.1.1" graphviz-rust = { version = "0.9.3", optional = true } -hashbrown = "0.15" itertools = { version = "0.14.0" } kuzu = {version = "=0.11.2", optional = true} nalgebra = { version = "0.33.2", optional = true } @@ -38,6 +37,8 @@ uuid = {version = "1.16.0", features = ["v4", "serde"]} schemars = { version = "1.1.0", features = ["chrono04", "uuid1"]} inventory = { version = "0.3", optional = true } csv = "1.4.0" +hashbrown = "0.17.1" +rustc-hash = "2.1.2" [dev-dependencies] criterion = { version = "0.5.1", features = ["html_reports"] } @@ -104,3 +105,7 @@ harness = false [[bench]] name = "load_events_mem" harness = false + +[[bench]] +name = "alignments" +harness = false diff --git a/process_mining/benches/alignments.rs b/process_mining/benches/alignments.rs new file mode 100644 index 00000000..72fc28af --- /dev/null +++ b/process_mining/benches/alignments.rs @@ -0,0 +1,47 @@ +//! Benchmark the time taken to compute optimal alignments of a log against a Petri net +use criterion::{black_box, criterion_group, criterion_main, Criterion}; +use process_mining::{ + conformance::alignments::{align_log, AlignmentOptions}, + core::event_data::case_centric::utils::activity_projection::{ + log_to_activity_projection, EventLogActivityProjection, + }, + test_utils::get_test_data_path, + EventLog, Importable, PetriNet, +}; +use std::time::Duration; + +fn load(log_name: &str, net_name: &str) -> (PetriNet, EventLogActivityProjection) { + let root = get_test_data_path(); + let log = EventLog::import_from_path(root.join("xes").join(log_name)).unwrap(); + let net = PetriNet::import_pnml(root.join("petri-net").join(net_name)).unwrap(); + (net, log_to_activity_projection(&log)) +} + +fn bench_alignments(c: &mut Criterion) { + let options = AlignmentOptions::default(); + + let (sepsis_net, sepsis_proj) = + load("Sepsis Cases - Event Log.xes.gz", "sepsis-DISCovered.apnml"); + let mut sepsis = c.benchmark_group("alignments"); + sepsis.sample_size(25); + sepsis.measurement_time(Duration::from_secs(20)); + sepsis.bench_function("sepsis", |b| { + b.iter(|| black_box(align_log(&sepsis_net, &sepsis_proj, &options))) + }); + sepsis.finish(); + + let (rtfm_net, rtfm_proj) = load( + "Road_Traffic_Fine_Management_Process.xes.gz", + "rtfm-imf-02.apnml", + ); + let mut rtfm = c.benchmark_group("alignments"); + rtfm.sample_size(10); + rtfm.measurement_time(Duration::from_secs(60)); + rtfm.bench_function("rtfm", |b| { + b.iter(|| black_box(align_log(&rtfm_net, &rtfm_proj, &options))) + }); + rtfm.finish(); +} + +criterion_group!(benches, bench_alignments); +criterion_main!(benches); diff --git a/process_mining/src/conformance/case_centric/alignments/cost.rs b/process_mining/src/conformance/case_centric/alignments/cost.rs new file mode 100644 index 00000000..f095ec77 --- /dev/null +++ b/process_mining/src/conformance/case_centric/alignments/cost.rs @@ -0,0 +1,31 @@ +//! Cost functions for alignments + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Cost function for alignment moves. +/// +/// Defines the cost of model moves, log moves, synchronous moves, and silent moves. +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct CostFunction { + /// Default cost for a model move (visible transition fires without matching log event) + pub model_move_cost: u32, + /// Default cost for a log move (log event not matched by model) + pub log_move_cost: u32, + /// Default cost for a synchronous move + pub sync_move_cost: u32, + /// Default cost for a silent/tau move + pub silent_move_cost: u32, +} + +impl CostFunction { + /// Standard cost function: model and log moves cost 1, sync and silent moves cost 0. + pub fn standard() -> Self { + Self { + model_move_cost: 1, + log_move_cost: 1, + sync_move_cost: 0, + silent_move_cost: 0, + } + } +} diff --git a/process_mining/src/conformance/case_centric/alignments/mod.rs b/process_mining/src/conformance/case_centric/alignments/mod.rs new file mode 100644 index 00000000..ddb8d7cc --- /dev/null +++ b/process_mining/src/conformance/case_centric/alignments/mod.rs @@ -0,0 +1,477 @@ +//! Implements optimal alignment search +use std::cell::RefCell; + +use macros_process_mining::register_binding; +use rayon::prelude::*; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::{ + conformance::alignments::{ + cost::CostFunction, + petri_net::{AlignmentError, PetriNetAlignmentSpace, PetriNetStep}, + sync_prod_net::SyncProductNet, + }, + core::{ + event_data::case_centric::utils::activity_projection::EventLogActivityProjection, + process_models::petri_net::TransitionID, + }, + utils::dijkstra_search::SearchState, + PetriNet, +}; + +pub mod cost; +pub mod petri_net; +pub mod sync_prod_net; + +/// A single alignment step +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub enum AlignmentMove { + /// Synchronous move (model and log agree) + SyncMove { + /// The transition that was fired + transition: TransitionID, + /// Index of the event in the trace + trace_event_index: usize, + }, + /// Model move (only the model moves,) + ModelMove { + /// The transition that was fired + transition: TransitionID, + }, + /// Log move (only the log moves) + LogMove { + /// Index of the event in the trace + trace_event_index: usize, + }, +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +/// Alignment Result +pub struct AlignmentResult { + /// The sequence of alignment moves + pub moves: Vec, + /// Total cost of the alignment + pub cost: u32, + /// Number of states visited during search + pub states_visited: usize, +} +/// Alignment result for a single trace variant +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct VariantAlignmentResult { + /// The variant's activity sequence + pub activities: Vec, + /// How many traces follow this variant + pub frequency: u64, + /// The alignment result or error for this variant + pub result: Result, +} + +/// Options for computing alignment +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct AlignmentOptions { + /// Cost function for alignment moves + pub cost_fn: cost::CostFunction, + /// Maximum number of states to visit before aborting (per trace). + /// `None` means no limit. + pub max_states: Option, +} +impl Default for AlignmentOptions { + fn default() -> Self { + Self { + cost_fn: CostFunction::standard(), + max_states: Some(100_000), + } + } +} +/// Compute alignments for all variants of an event log. +/// +/// Permits at most 255 ([`petri_net::TokenCount::MAX`]) tokens in each place. +pub fn align_log<'a>( + net: &PetriNet, + log: impl Into<&'a EventLogActivityProjection>, + options: &AlignmentOptions, +) -> Vec { + let projection: &EventLogActivityProjection = log.into(); + align_variants(net, projection, options) +} + +/// Compute alignments for all variants from a pre-computed activity projection. +/// +/// Permits at most 255 ([`petri_net::TokenCount::MAX`]) tokens in each place. +#[register_binding] +pub fn align_variants( + net: &PetriNet, + projection: &EventLogActivityProjection, + #[bind(default)] options: &AlignmentOptions, +) -> Vec { + projection + .traces + .par_iter() + .map(|(trace_indices, count)| { + let x: Vec<_> = trace_indices + .iter() + .map(|i| projection.activities[*i].as_str()) + .collect(); + let sp = SyncProductNet::construct(net, &x, &options.cost_fn); + + thread_local! { + static CTX: RefCell<(PetriNetAlignmentSpace, SearchState)> = + RefCell::new((PetriNetAlignmentSpace::default(), SearchState::default())); + } + let activities: Vec = trace_indices + .iter() + .map(|&idx| projection.activities[idx].clone()) + .collect(); + let result = CTX.with(|ctx| { + let (space, state) = &mut *ctx.borrow_mut(); + sp.map_err(AlignmentError::SyncProdNetConstructionFailed) + .and_then(|sp| petri_net::align(&sp, space, state, options.max_states)) + }); + VariantAlignmentResult { + activities, + frequency: *count, + result, + } + }) + .collect() +} + +/// Compute alignment for a single trace (given as activity sequence). +/// +/// Permits at most 255 ([`petri_net::TokenCount::MAX`]) tokens in each place. +pub fn align_trace( + net: &PetriNet, + trace: &[&str], + options: &AlignmentOptions, +) -> Result { + let sp = SyncProductNet::construct(net, trace, &options.cost_fn)?; + petri_net::align( + &sp, + &mut PetriNetAlignmentSpace::default(), + &mut SearchState::default(), + options.max_states, + ) +} + +/// Compute alignment for a single trace (given as activity sequence). +/// +/// Permits at most 255 ([`petri_net::TokenCount::MAX`]) tokens in each place. +#[allow(dead_code)] +#[register_binding(stringify_error, name = "align_trace")] +fn align_trace_binding( + net: &PetriNet, + trace: &[String], + #[bind(default)] options: &AlignmentOptions, +) -> Result { + let trace_as_str: Vec<_> = trace.iter().map(|s| s.as_str()).collect(); + align_trace(net, &trace_as_str, options) +} + +/// Align the empty trace to the given model +/// with the specified options +/// +/// Permits at most 255 ([`petri_net::TokenCount::MAX`]) tokens in each place. +#[register_binding(stringify_error)] +pub fn align_empty_trace( + net: &PetriNet, + #[bind(default)] options: &AlignmentOptions, +) -> Result { + let sp = SyncProductNet::construct(net, &[], &options.cost_fn)?; + petri_net::align( + &sp, + &mut PetriNetAlignmentSpace::default(), + &mut SearchState::default(), + options.max_states, + ) +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +/// Alignment Fitness Result +pub struct FitnessResult { + /// Log fitness, as the total computed fitness (summing up the costs for all traces) + pub log_fitness: f64, + /// Average trace fitness (across all traces) + pub average_fitness: f64, + /// Fraction of traces that perfectly fit (i.e., have an alignment cost of `0`) + pub perfectly_fitting_frac: f64, + /// The total cost, summed up from all traces + pub total_costs: u64, +} + +/// Compute fitness stats from alignment results +/// +/// Also constructs the empty-trace alignment (shortest path through model) +#[register_binding(stringify_error)] +pub fn compute_fitness( + align_res: &[VariantAlignmentResult], + net: &PetriNet, + #[bind(default)] options: &AlignmentOptions, +) -> Result { + let empty = align_empty_trace(net, options)?; + let model_path_min = empty.cost; + let mut num_perfectly_fitting = 0; + let mut total_costs = 0; + let mut fitness_sum_for_avg = 0f64; + let mut num_traces = 0; + let mut num_events = 0; + for variant in align_res { + let res = variant.result.as_ref().map_err(|e| e.clone())?; + let costs = res.cost; + if costs == 0 { + num_perfectly_fitting += variant.frequency; + } + total_costs += variant.frequency * costs as u64; + num_traces += variant.frequency; + num_events += variant.frequency * variant.activities.len() as u64; + let denom = variant.activities.len() as f64 * options.cost_fn.log_move_cost as f64 + + model_path_min as f64; + // denom == 0 means an empty trace against a net with initial == final marking: Perfectly fitting + let fitness = if denom == 0.0 { + 1f64 + } else { + 1f64 - (costs as f64 / denom) + }; + fitness_sum_for_avg += variant.frequency as f64 * fitness; + } + let log_denom = num_events as f64 * options.cost_fn.log_move_cost as f64 + + num_traces as f64 * model_path_min as f64; + let log_fitness = if log_denom == 0.0 { + 1f64 + } else { + 1f64 - (total_costs as f64 / log_denom) + }; + Ok(FitnessResult { + log_fitness, + average_fitness: if num_traces == 0 { + // Could be either way.. + 0f64 + } else { + fitness_sum_for_avg / num_traces as f64 + }, + perfectly_fitting_frac: if num_traces == 0 { + 0f64 + } else { + num_perfectly_fitting as f64 / num_traces as f64 + }, + total_costs, + }) +} + +#[cfg(test)] +mod test { + use std::{collections::HashSet, time::Instant}; + + use crate::{ + conformance::alignments::{ + align_empty_trace, align_log, compute_fitness, + cost::CostFunction, + petri_net::AlignmentError, + sync_prod_net::{SyncProdNetConstructionError, SyncProductNet}, + AlignmentOptions, + }, + core::{ + event_data::case_centric::utils::activity_projection::log_to_activity_projection, + process_models::petri_net::{ArcType, PlaceID}, + }, + test_utils::get_test_data_path, + utils::dijkstra_search::SearchError, + EventLog, Importable, PetriNet, + }; + + fn align_helper( + log_name: &str, + net_name: &str, + ) -> ( + Vec, + Result, + ) { + let test_path = get_test_data_path(); + let log = EventLog::import_from_path(test_path.join("xes").join(log_name)).unwrap(); + let net = PetriNet::import_pnml(test_path.join("petri-net").join(net_name)).unwrap(); + let act_proj = log_to_activity_projection(&log); + let options = AlignmentOptions::default(); + let now = Instant::now(); + let result = align_log(&net, &act_proj, &options); + println!("Aligning traces took {:?}", now.elapsed()); + let fitness = compute_fitness(&result, &net, &options); + println!("{fitness:?}"); + (result, fitness) + } + + #[test] + fn sepsis_total_cost() { + let (_alignment, fitness) = + align_helper("Sepsis Cases - Event Log.xes.gz", "sepsis-DISCovered.apnml"); + let fitness = fitness.unwrap(); + // Ground truth total alignment cost was computed and additionally verified with external source (PM4Py) + assert_eq!(fitness.total_costs, 4118); + } + + #[test] + fn rtfm_total_cost() { + let (_alignment, fitness) = align_helper( + "Road_Traffic_Fine_Management_Process.xes.gz", + "rtfm-imf-02.apnml", + ); + let fitness = fitness.unwrap(); + // Ground truth total alignment cost was computed and additionally verified with external source (PM4Py) + assert_eq!(fitness.total_costs, 17650); + } + + #[test] + fn no_initial_marking_err() { + let test_path = get_test_data_path(); + let mut net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + net.initial_marking = None; + let sn = SyncProductNet::construct(&net, &[], &CostFunction::standard()); + assert_eq!(sn, Err(SyncProdNetConstructionError::NoInitialMarking)); + } + #[test] + fn no_final_markings_err() { + let test_path = get_test_data_path(); + let mut net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + net.final_markings = None; + let sn = SyncProductNet::construct(&net, &[], &CostFunction::standard()); + assert_eq!(sn, Err(SyncProdNetConstructionError::NoFinalMarking)); + } + #[test] + fn unknown_place_in_initial_marking_err() { + let test_path = get_test_data_path(); + let mut net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + let new_id = PlaceID(uuid::Uuid::new_v4()); + net.initial_marking + .as_mut() + .expect("exists in apnml") + .insert(new_id, 1); + let sn = SyncProductNet::construct(&net, &[], &CostFunction::standard()); + assert_eq!( + sn, + Err(SyncProdNetConstructionError::InvalidPlaceInMarking(new_id)) + ); + } + #[test] + fn unknown_place_in_final_marking_err() { + let test_path = get_test_data_path(); + let mut net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + let new_id = PlaceID(uuid::Uuid::new_v4()); + net.final_markings + .as_mut() + .expect("exists in apnml") + .first_mut() + .expect("one final marking exists") + .insert(new_id, 1); + let sn = SyncProductNet::construct(&net, &[], &CostFunction::standard()); + assert_eq!( + sn, + Err(SyncProdNetConstructionError::InvalidPlaceInMarking(new_id)) + ); + } + + #[test] + fn final_marking_unreachable_err() { + let test_path = get_test_data_path(); + let log = EventLog::import_from_path( + test_path + .join("xes") + .join("Sepsis Cases - Event Log.xes.gz"), + ) + .unwrap(); + let mut net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + let places_in_final_marking: HashSet<_> = net + .final_markings + .as_mut() + .expect("exists in file") + .first_mut() + .expect("not empty") + .keys() + .map(|id| id.0) + .collect(); + net.arcs.retain(|arc| match arc.from_to { + ArcType::PlaceTransition(_, _) => true, + ArcType::TransitionPlace(_, place) => !places_in_final_marking.contains(&place), + }); + let act_proj = log_to_activity_projection(&log); + let options = AlignmentOptions { + cost_fn: CostFunction::standard(), + max_states: None, + }; + let empty_trace_align = align_empty_trace(&net, &options); + assert_eq!( + empty_trace_align, + Err(AlignmentError::SearchError(SearchError::Unreachable)) + ); + let result = align_log(&net, &act_proj, &options); + for variant in result { + assert_eq!( + variant.result, + Err(AlignmentError::SearchError(SearchError::Unreachable)) + ); + } + } + #[test] + fn max_states_reached_err() { + let test_path = get_test_data_path(); + let log = EventLog::import_from_path( + test_path + .join("xes") + .join("Sepsis Cases - Event Log.xes.gz"), + ) + .unwrap(); + let net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-DISCovered.apnml")) + .unwrap(); + let act_proj = log_to_activity_projection(&log); + let options = AlignmentOptions { + cost_fn: CostFunction::standard(), + max_states: Some(10), + }; + let empty_trace_align = align_empty_trace(&net, &options); + assert_eq!( + empty_trace_align, + Err(AlignmentError::SearchError(SearchError::LimitReached)) + ); + let result = align_log(&net, &act_proj, &options); + for variant in result { + assert_eq!( + variant.result, + Err(AlignmentError::SearchError(SearchError::LimitReached)) + ); + } + } + #[test] + fn max_states_reached_not_easy_sound_err() { + let test_path = get_test_data_path(); + let log = EventLog::import_from_path( + test_path + .join("xes") + .join("Sepsis Cases - Event Log.xes.gz"), + ) + .unwrap(); + let net = + PetriNet::import_pnml(test_path.join("petri-net").join("sepsis-fodina.apnml")).unwrap(); + let act_proj = log_to_activity_projection(&log); + let options = AlignmentOptions::default(); + let empty_trace_align = align_empty_trace(&net, &options); + assert_eq!( + empty_trace_align, + Err(AlignmentError::SearchError(SearchError::LimitReached)) + ); + let result = align_log(&net, &act_proj, &options); + for variant in result { + assert_eq!( + variant.result, + Err(AlignmentError::SearchError(SearchError::LimitReached)) + ); + } + } +} diff --git a/process_mining/src/conformance/case_centric/alignments/petri_net.rs b/process_mining/src/conformance/case_centric/alignments/petri_net.rs new file mode 100644 index 00000000..51c7560c --- /dev/null +++ b/process_mining/src/conformance/case_centric/alignments/petri_net.rs @@ -0,0 +1,290 @@ +//! Optimal alignment search: the synchronous product net as a [`SearchProblem`], solved with the +//! generic [`crate::utils::dijkstra_search`] Dijkstra. +use std::hash::Hasher; + +use hashbrown::HashTable; +use rustc_hash::FxHasher; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::{ + conformance::alignments::{ + sync_prod_net::{SyncProdNetConstructionError, SyncProdNetTransition, SyncProductNet}, + AlignmentMove, AlignmentResult, + }, + utils::dijkstra_search::{search, NodeID, SearchError, SearchProblem, SearchState}, +}; + +/// Alignment Error +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +pub enum AlignmentError { + /// Search Error + SearchError(SearchError), + /// Constructing the synchronous product net failed + SyncProdNetConstructionFailed(SyncProdNetConstructionError), +} + +impl std::fmt::Display for AlignmentError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{self:?}") + } +} +impl std::error::Error for AlignmentError {} + +impl From for AlignmentError { + fn from(value: SyncProdNetConstructionError) -> Self { + Self::SyncProdNetConstructionFailed(value) + } +} + +impl From for AlignmentError { + fn from(value: SearchError) -> Self { + Self::SearchError(value) + } +} + +/// Type representing the count of tokens (e.g., in a marking) +/// +/// May be changed to larger types if Petri nets with more than 255 tokens in a place should be supported. +pub type TokenCount = u8; + +/// Type representing trace position +pub type TracePos = u16; + +/// An edge/step of the Petri-net search: the transition fired, and whether it is a log move. +/// +/// The log-move-flag allows for log-before-model-moves pruning in [`PetriNetAlignment::expand`]. +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub(crate) struct PetriNetStep { + /// Index of the fired transition in the sync. prod. net + transition: u32, + /// Whether the fired transition is a log move + log_move: bool, +} + +/// Reusable state storage for a Petri-net alignment search +#[derive(Debug, Default)] +pub(crate) struct PetriNetAlignmentSpace { + /// Number of model places + num_places: usize, + /// Flat storage for model markings, indexed using [`NodeID`] + markings: Vec, + /// Trace position per node, indexed using [`NodeID`] + trace_pos: Vec, + /// Index of visited states, mapping a `(marking, trace_pos)` tuple to a [`NodeID`] + seen: HashTable, + /// Current marking (re-used to reduce allocations) + current: Vec, + /// Next marking, reached by firing a transition (re-used to reduce allocations) + next: Vec, +} + +impl PetriNetAlignmentSpace { + fn reset(&mut self, net: &SyncProductNet) { + self.num_places = net.num_model_places; + self.markings.clear(); + self.trace_pos.clear(); + self.seen.clear(); + self.current.resize(net.num_model_places, 0); + self.next.resize(net.num_model_places, 0); + self.markings + .extend_from_slice(&net.initial_marking[..net.num_model_places]); + self.trace_pos.push(0); + self.add_seen(0); + } + + #[inline] + fn add_seen(&mut self, node: NodeID) { + let off = node as usize * self.num_places; + let hash = hash_state( + &self.markings[off..off + self.num_places], + self.trace_pos[node as usize], + ); + let markings = &self.markings; + let trace_pos = &self.trace_pos; + let num_places = self.num_places; + self.seen.insert_unique(hash, node, |other| { + let off = *other as usize * num_places; + hash_state(&markings[off..off + num_places], trace_pos[*other as usize]) + }); + } + + #[inline] + fn find_seen(&self, marking: &[TokenCount], trace_position: TracePos) -> Option { + let hash = hash_state(marking, trace_position); + let num_places = self.num_places; + self.seen + .find(hash, |node| { + let off = *node as usize * num_places; + &self.markings[off..off + num_places] == marking + && self.trace_pos[*node as usize] == trace_position + }) + .copied() + } +} + +/// Alignment as a [`SearchProblem`]: a state is a `(model marking, trace position)`, an edge fires +/// a sync. prod. net transition +#[derive(Debug)] +struct PetriNetAlignment<'a> { + net: &'a SyncProductNet, + space: &'a mut PetriNetAlignmentSpace, +} + +impl<'a> PetriNetAlignment<'a> { + /// Build the alignment problem over `net`, storing search state in `space` + fn new(net: &'a SyncProductNet, space: &'a mut PetriNetAlignmentSpace) -> Self { + Self { net, space } + } +} + +/// Instantiate the search problem ([`SearchProblem`]) for the Petri net alignment +impl SearchProblem for PetriNetAlignment<'_> { + /// A single step in the search corresponds to firing a sync net transition, represented by [`PetriNetStep`]. + type Step = PetriNetStep; + /// Costs are represented as `u32`, also see [`super::cost::CostFunction`] + type Cost = u32; + + /// Reset the search space and return the initial node ID (0) + fn initial(&mut self) -> NodeID { + self.space.reset(self.net); + 0 + } + /// Get maximal edge cost in the synchronous product net (needed for bucket setup) + fn max_edge_cost(&self) -> u32 { + self.net.max_edge_cost + } + + /// Check whether the given node is a goal state (final marking and trace position at the end of the trace) + #[inline] + fn is_goal(&self, node: NodeID) -> bool { + let np = self.net.num_model_places; + let off = node as usize * np; + self.space.trace_pos[node as usize] == self.net.trace_length + && self.space.markings[off..off + np] == self.net.final_marking[..np] + } + + /// Expand the given node, emitting all reachable nodes and their costs via the `emit` callback + #[inline] + fn expand( + &mut self, + node: NodeID, + via: Option, + mut emit: F, + ) { + let net = self.net; + let space = &mut *self.space; + let np = space.num_places; + let off = node as usize * np; + let trace_pos = space.trace_pos[node as usize]; + let last_move_was_log = via.is_some_and(|s| s.log_move); + space + .current + .copy_from_slice(&space.markings[off..off + np]); + + // Log/sync moves for the current event, then model moves (fixed ordering prunes states). + let log_or_sync = net + .transitions_by_trace_pos + .get(trace_pos as usize) + .map(|v| v.as_slice()) + .unwrap_or_default(); + // After a log move, model moves are pruned, so the range collapses to empty. + let model_end = if last_move_was_log { + 0 + } else { + net.num_model_trans + }; + + for trans_idx in log_or_sync.iter().copied().chain(0..model_end) { + let trans = &net.transitions[trans_idx]; + if !is_enabled(&space.current, trans) { + continue; + } + if fire_transition(&space.current, &mut space.next, trans).is_none() { + continue; + } + let is_model_move = matches!(trans.move_type, AlignmentMove::ModelMove { .. }); + let new_trace_pos = if is_model_move { + trace_pos + } else { + trace_pos + 1 + }; + let step = PetriNetStep { + transition: trans_idx as u32, + log_move: matches!(trans.move_type, AlignmentMove::LogMove { .. }), + }; + + let cost = trans.cost; + match space.find_seen(&space.next, new_trace_pos) { + Some(existing) => emit(existing, false, cost, step), + None => { + let new_id = space.trace_pos.len() as NodeID; + space.markings.extend_from_slice(&space.next); + space.trace_pos.push(new_trace_pos); + space.add_seen(new_id); + emit(new_id, true, cost, step); + } + } + } + } +} + +/// Compute an optimal alignment using [`search`] +pub(crate) fn align( + net: &SyncProductNet, + space: &mut PetriNetAlignmentSpace, + state: &mut SearchState, + max_states: Option, +) -> Result { + let mut problem = PetriNetAlignment::new(net, space); + let res = search(&mut problem, state, max_states)?; + Ok(AlignmentResult { + moves: res + .path + .iter() + .map(|s| net.transitions[s.transition as usize].move_type.clone()) + .collect(), + cost: res.cost, + states_visited: res.states_visited, + }) +} + +#[inline] +/// Tests whether the given transition is enabled in the given marking +fn is_enabled(marking: &[TokenCount], trans: &SyncProdNetTransition) -> bool { + trans + .inputs + .iter() + .all(|(place, weight)| &marking[*place] >= weight) +} + +#[inline] +#[must_use] +/// Fire the given transition, transforming the current marking into the `reached` marking. +/// +/// Returns `None` if the reached marking would exceed `TokenCount::MAX` tokens. +/// In this case, the reached marking is considered out-of-bounds and should be pruned. +fn fire_transition( + current: &[TokenCount], + reached: &mut [TokenCount], + trans: &SyncProdNetTransition, +) -> Option<()> { + reached.copy_from_slice(current); + for (place, weight) in &trans.inputs { + reached[*place] -= weight; + } + for (place, weight) in &trans.outputs { + // Handle overflow to prevent arriving at incorrect markings + reached[*place] = reached[*place].checked_add(*weight)?; + } + Some(()) +} + +#[inline] +/// Hash a given state (combination of marking and trace position) +fn hash_state(marking: &[TokenCount], trace_pos: TracePos) -> u64 { + let mut h = FxHasher::default(); + h.write(marking); + h.write_u16(trace_pos); + h.finish() +} diff --git a/process_mining/src/conformance/case_centric/alignments/sync_prod_net.rs b/process_mining/src/conformance/case_centric/alignments/sync_prod_net.rs new file mode 100644 index 00000000..a909b37e --- /dev/null +++ b/process_mining/src/conformance/case_centric/alignments/sync_prod_net.rs @@ -0,0 +1,215 @@ +//! Synchronous product net used for efficiently computing alignments + +use std::collections::HashMap; + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::{ + conformance::alignments::{ + cost::CostFunction, + petri_net::{TokenCount, TracePos}, + AlignmentMove, + }, + core::process_models::petri_net::{ArcType, PlaceID, TransitionID}, + PetriNet, +}; + +/// A transition in the synchronous product net +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct SyncProdNetTransition { + /// The move this transition represents (model-transition / trace-event indices) + pub(crate) move_type: AlignmentMove, + /// The pre-computed cost of firing this transition + pub(crate) cost: u32, + /// Incoming places (`place_index`, weight), i.e., which token to consume + pub(crate) inputs: Vec<(usize, TokenCount)>, + /// Outgoing places (`place_index`, weight), i.e., which token to produce + pub(crate) outputs: Vec<(usize, TokenCount)>, +} + +/// The synchronous product of a Petri net and a trace. +/// +/// Only model places exist; the trace position is tracked additionally in the search +/// Transitions: [`model_moves` for model transitions, `log_moves`, `sync_moves`] +#[derive(Debug, PartialEq)] +pub(crate) struct SyncProductNet { + /// Number of model places + pub(crate) num_model_places: usize, + /// Length of the trace + pub(crate) trace_length: TracePos, + /// All transitions in the sync product + pub(crate) transitions: Vec, + /// Initial marking (tokens per place) + pub(crate) initial_marking: Vec, + /// Final marking (tokens per place) + pub(crate) final_marking: Vec, + /// Log/sync transition indices grouped by trace position. + /// `transitions_by_trace_pos[r]` holds the log/sync transitions for event `r` + /// (fireable once r trace events have been advanced). + pub(crate) transitions_by_trace_pos: Vec>, + /// Number of model/silent transitions (fireable at any rank); they occupy indices `0..n`. + pub(crate) num_model_trans: usize, + /// Largest cost over all transitions (precomputed for the search's bucket sizing) + pub(crate) max_edge_cost: u32, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +/// Error when constructing the sync product net +pub enum SyncProdNetConstructionError { + /// A unknown place id was referenced in a marking + InvalidPlaceInMarking(PlaceID), + /// No final marking found + NoFinalMarking, + /// No initial marking found + NoInitialMarking, +} +impl std::fmt::Display for SyncProdNetConstructionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{self:?}") + } +} +impl std::error::Error for SyncProdNetConstructionError {} + +impl SyncProductNet { + /// Construct the sync product net from a Petri net and trace + pub(crate) fn construct( + net: &PetriNet, + trace: &[&str], + cost_fn: &CostFunction, + ) -> Result { + let mut transitions = Vec::new(); + let model_place_map: HashMap<&Uuid, usize> = net + .places + .iter() + .enumerate() + .map(|(i, p)| (p.0, i)) + .collect(); + // All sync-net places are model places + let num_model_places = net.places.len(); + let mut model_trans_map: HashMap<&Uuid, usize> = HashMap::new(); + let mut model_trans_with_label: HashMap<&str, Vec> = HashMap::new(); + let num_model_trans = net.transitions.len(); + let mut transitions_by_rank = vec![vec![]; trace.len()]; + // Model moves + for (trans_id, trans) in &net.transitions { + let model_idx = transitions.len(); + model_trans_map.insert(trans_id, model_idx); + if let Some(label) = &trans.label { + model_trans_with_label + .entry(label) + .or_default() + .push(model_idx); + } + let inputs = net + .arcs + .iter() + .filter_map(|arc| match arc.from_to { + ArcType::PlaceTransition(x, to) if &to == trans_id => { + Some((*model_place_map.get(&x)?, arc.weight as TokenCount)) + } + _ => None, + }) + .collect(); + let outputs = net + .arcs + .iter() + .filter_map(|arc| match arc.from_to { + ArcType::TransitionPlace(from, x) if &from == trans_id => { + Some((*model_place_map.get(&x)?, arc.weight as TokenCount)) + } + _ => None, + }) + .collect(); + transitions.push(SyncProdNetTransition { + move_type: AlignmentMove::ModelMove { + transition: TransitionID(*trans_id), + }, + cost: if trans.label.is_none() { + cost_fn.silent_move_cost + } else { + cost_fn.model_move_cost + }, + inputs, + outputs, + }); + } + // Log + Sync moves + // There are no log places, instead the trace position is tracked by the + // searcher (using trace_pos) A log move has no model effect; sync move uses the model + // transition's preset/postset. + for (index, activity) in trace.iter().enumerate() { + transitions_by_rank[index].push(transitions.len()); + + // Log moves + transitions.push(SyncProdNetTransition { + move_type: AlignmentMove::LogMove { + trace_event_index: index, + }, + cost: cost_fn.log_move_cost, + inputs: vec![], + outputs: vec![], + }); + // Sync moves + let model_trans_that_can_sync = model_trans_with_label.get(activity); + for s in model_trans_that_can_sync.into_iter().flatten() { + transitions_by_rank[index].push(transitions.len()); + let trans_id = match transitions[*s].move_type { + AlignmentMove::ModelMove { transition } => transition, + _ => unreachable!("Has to be a model move"), + }; + transitions.push(SyncProdNetTransition { + move_type: AlignmentMove::SyncMove { + transition: trans_id, + trace_event_index: index, + }, + cost: cost_fn.sync_move_cost, + inputs: transitions[*s].inputs.clone(), + outputs: transitions[*s].outputs.clone(), + }); + } + } + + // Build initial marking + let mut initial_marking: Vec = vec![0; num_model_places]; + // Initial marking for model + let im = net + .initial_marking + .as_ref() + .ok_or(SyncProdNetConstructionError::NoInitialMarking)?; + for (place_id, count) in im { + let index = model_place_map.get(&place_id.0).ok_or( + SyncProdNetConstructionError::InvalidPlaceInMarking(*place_id), + )?; + initial_marking[*index] = *count as TokenCount; + } + + // Build final marking + let mut final_marking: Vec = vec![0; num_model_places]; + // Final marking for model + let fm = net + .final_markings + .as_ref() + .and_then(move |f| f.first()) + .ok_or(SyncProdNetConstructionError::NoFinalMarking)?; + for (place_id, count) in fm { + let index = model_place_map.get(&place_id.0).ok_or( + SyncProdNetConstructionError::InvalidPlaceInMarking(*place_id), + )?; + final_marking[*index] = *count as TokenCount; + } + + let max_edge_cost = transitions.iter().map(|t| t.cost).max().unwrap_or(1); + Ok(Self { + num_model_places, + trace_length: trace.len() as TracePos, + transitions, + initial_marking, + final_marking, + transitions_by_trace_pos: transitions_by_rank, + num_model_trans, + max_edge_cost, + }) + } +} diff --git a/process_mining/src/conformance/case_centric/mod.rs b/process_mining/src/conformance/case_centric/mod.rs index cc1b2c43..39b084e5 100644 --- a/process_mining/src/conformance/case_centric/mod.rs +++ b/process_mining/src/conformance/case_centric/mod.rs @@ -1,3 +1,5 @@ //! Case-centric Conformance-checking Approaches #[cfg(feature = "token-based-replay")] pub mod token_based_replay; + +pub mod alignments; diff --git a/process_mining/src/core/process_models/case_centric/petri_net/petri_net_struct.rs b/process_mining/src/core/process_models/case_centric/petri_net/petri_net_struct.rs index 7c1ac321..8cf71667 100644 --- a/process_mining/src/core/process_models/case_centric/petri_net/petri_net_struct.rs +++ b/process_mining/src/core/process_models/case_centric/petri_net/petri_net_struct.rs @@ -102,7 +102,9 @@ impl From<&Place> for PlaceID { } } -#[derive(Debug, PartialEq, Clone, Copy, Serialize, Deserialize, Hash, Eq, PartialOrd, Ord)] +#[derive( + Debug, PartialEq, Clone, Copy, Serialize, Deserialize, Hash, Eq, PartialOrd, Ord, JsonSchema, +)] /// Transition ID pub struct TransitionID(pub Uuid); diff --git a/process_mining/src/lib.rs b/process_mining/src/lib.rs index 23c5cb65..8e071534 100644 --- a/process_mining/src/lib.rs +++ b/process_mining/src/lib.rs @@ -12,6 +12,7 @@ pub mod analysis; pub mod conformance; pub mod core; pub mod discovery; +pub mod utils; pub use core::io::{Exportable, Importable}; diff --git a/process_mining/src/utils/dijkstra_search.rs b/process_mining/src/utils/dijkstra_search.rs new file mode 100644 index 00000000..876059d5 --- /dev/null +++ b/process_mining/src/utils/dijkstra_search.rs @@ -0,0 +1,231 @@ +//! Generic Dijkstra state space exploration +//! +//! A [`SearchProblem`] defines states, edges, and the goal state. +//! The [`search`] function finds an optimal path in the statespace. +//! Consumers (e.g. alignments) implement [`SearchProblem`] for their specific +//! state space and can then reuse [`search`]. +use std::{ + collections::VecDeque, + ops::{Add, Rem}, +}; + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Index of a node in the search (assigned in creation order). +/// +/// Must be dense, i.e., no index may be skipped. +pub type NodeID = u32; + +/// A state space with non-negative integer edge costs, explored by [`search`]. +/// +/// The problem instance owns the state storage, while the search tracks distance, +/// parent and [`Step`] per node, keyed by [`NodeID`]. +/// +/// New nodes get ids in creation order (see [`expand`]), without skipping an index. +/// +/// [`Step`]: SearchProblem::Step +/// [`expand`]: SearchProblem::expand +pub trait SearchProblem { + /// The edge taken to reach a node, used to reconstruct paths. + type Step: Copy + Default; + + /// The integer type for edge and path costs. + /// Any primitive integer (`u16`, `u32`, etc.) may be used. + /// The type must be large enough to hold the largest path costs. + type Cost: Copy + + Ord + + Add + + Rem + + From + + TryInto; + + /// The start state; returns its id (`0`). + fn initial(&mut self) -> NodeID; + + /// Largest possible cost of a _single_ edge. + /// + /// This determines the number of buckets (`num_buckets` = `max_edge_cost` + 1). + fn max_edge_cost(&self) -> Self::Cost; + + /// Whether the given `node` is a final (goal) state. + fn is_goal(&self, node: NodeID) -> bool; + + /// Generate the successors of `node`. + /// For each outgoing edge, find or create the node id for the + /// reached state and call `emit(successor, is_new, edge_cost, step)`: + /// `is_new` is `true` when the state was newly created, and a new id must + /// equal the current node count (no id may be skipped). + /// `via` is the edge `node` is currently best reached by (`None` for the start) + fn expand( + &mut self, + node: NodeID, + via: Option, + emit: F, + ); +} + +#[derive(Debug)] +struct Node { + /// Shortest known distance to this node from the start node + distance: C, + /// The parent node (from which this one is reached). + /// + /// For the start node, this is set to its own + parent: NodeID, + /// The edge fired to reach this node from its parent + step: S, + /// If this node is finished (final distance known and removed from the queue) + finished: bool, +} + +/// An optimal path found by [`search`] +#[derive(Debug, Clone)] +pub struct SearchResult { + /// The edges fired from start to goal, in order + pub path: Vec, + /// Total cost of the path + pub cost: C, + /// Number of states visited during search + pub states_visited: usize, +} + +/// Reason [`search`] found no path +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +pub enum SearchError { + /// The specified maximum number of states was reached + LimitReached, + /// No goal state is reachable + Unreachable, + /// The maximum edge cost does not fit `usize`, so the bucket queue cannot be sized + MaxEdgeCostTooLarge, +} + +/// Reusable node store and bucket queue for [`search`], cleared initially. +/// Reusing it across searches avoids reallocations. +#[derive(Debug)] +pub struct SearchState { + /// Node info, indexed using [`NodeID`] + nodes: Vec>, + /// Priority bucket queue. + /// + /// Nodes with distance `d` are scheduled in `buckets[d % len]`, + /// where `len` is the number of buckets (max edge cost + 1) + buckets: Vec>, +} + +impl Default for SearchState { + fn default() -> Self { + Self { + nodes: Vec::new(), + buckets: Vec::new(), + } + } +} + +/// Search for an optimal path from the start to a goal state using Dijkstra +#[inline] +pub fn search( + problem: &mut P, + state: &mut SearchState, + max_states: Option, +) -> Result, SearchError> { + let num_buckets_cost = problem.max_edge_cost() + P::Cost::from(1); + let num_buckets: usize = num_buckets_cost + .try_into() + .map_err(|_| SearchError::MaxEdgeCostTooLarge)?; + let start = problem.initial(); + + let nodes = &mut state.nodes; + let buckets = &mut state.buckets; + nodes.clear(); + nodes.push(Node { + distance: P::Cost::from(0), + parent: start, + step: P::Step::default(), + finished: false, + }); + buckets.resize_with(num_buckets, VecDeque::new); + buckets.iter_mut().for_each(VecDeque::clear); + buckets[0].push_back(start); + + let limit = max_states.unwrap_or(usize::MAX); + let mut queued: usize = 1; + let mut bucket: usize = 0; + let mut states_visited: usize = 0; + + while queued > 0 { + while buckets[bucket].is_empty() { + bucket = (bucket + 1) % num_buckets; + } + let node_id = buckets[bucket].pop_front().expect("Bucket not empty"); + queued -= 1; + + if nodes[node_id as usize].finished { + continue; + } + nodes[node_id as usize].finished = true; + states_visited += 1; + if states_visited > limit { + return Err(SearchError::LimitReached); + } + + if problem.is_goal(node_id) { + return Ok(SearchResult { + path: reconstruct(nodes, node_id), + cost: nodes[node_id as usize].distance, + states_visited, + }); + } + + let distance = nodes[node_id as usize].distance; + let via = if nodes[node_id as usize].parent == node_id { + None + } else { + Some(nodes[node_id as usize].step) + }; + problem.expand(node_id, via, |next_id, is_new, cost, step| { + let new_distance = distance + cost; + // Modulo in cost space first: the remainder is `< num_buckets` so in usize. + let idx = (new_distance % num_buckets_cost).try_into().unwrap_or(0); + if is_new { + debug_assert_eq!(next_id as usize, nodes.len(), "no NodeID should be skipped"); + nodes.push(Node { + distance: new_distance, + parent: node_id, + step, + finished: false, + }); + // add it to the back of corresponding bucket + buckets[idx].push_back(next_id); + queued += 1; + } else { + let node = &mut nodes[next_id as usize]; + // node already existed before, so update it if distance is smaller. + if !node.finished && new_distance < node.distance { + node.distance = new_distance; + node.parent = node_id; + node.step = step; + buckets[idx].push_back(next_id); + queued += 1; + } + } + }); + } + Err(SearchError::Unreachable) +} + +/// Reconstruct a path using the steps taken +/// +/// Returns an ordered sequence of steps to reach a goal node from the start node. +fn reconstruct(nodes: &[Node], target: NodeID) -> Vec { + let mut path = Vec::new(); + let mut current = target; + while nodes[current as usize].parent != current { + let node = &nodes[current as usize]; + path.push(node.step); + current = node.parent; + } + path.reverse(); + path +} diff --git a/process_mining/src/utils/mod.rs b/process_mining/src/utils/mod.rs new file mode 100644 index 00000000..62beb042 --- /dev/null +++ b/process_mining/src/utils/mod.rs @@ -0,0 +1,2 @@ +//! Cross-cutting utilities not tied to a specific process mining technique +pub mod dijkstra_search;