Merge branch 'main' of https://git.rwth-aachen.de/jakob.junck/super-variant-explorer

src/backend/mine_log.py

+from fastapi import FastAPI, UploadFile, File
+from pydantic import BaseModel
+from typing import List, Dict, Any, Optional
+import uvicorn
+
+from ocpa.objects.log.importer.ocel import factory as ocel_import_factory
+from ocpa.visualization.log.variants import factory as variants_visualization_factory
+from ocpa.algo.util.filtering.log import case_filtering
+from ocpa.objects.log.exporter.ocel import factory as ocel_export_factory
+
+import ocsv.Input_Extraction_Definition as IED  # Seems like a helper class to define data structures and querying lanes
+import ocsv.Super_Variant_Definition as SVD     # Super Variant Definition, super lane definition
+import ocsv.Super_Variant_Visualization as SVV  # Visualization of super variants
+import ocsv.Intra_Variant_Summarization as IAVS # 
+import ocsv.Summarization_Selection as SS
+import ocsv.Intra_Variant_Generation as IAVG
+import ocsv.Inter_Variant_Summarization as IEVS
+import ocsv.Inter_Variant_Generation as IEVG
+import ocsv.Super_Variant_Hierarchy as SVH
+import time 
+import numpy as np
+
+app = FastAPI()
+
+class Parameters(BaseModel):
+    execution_extraction: Optional[str] = "leading_type"
+    leading_type: Optional[str] = "application"
+    max_levels: Optional[int] = 4
+    frequency_distribution_type: Optional[str] = "NORMAL"
+
+
+@app.post("/process_ocel/")
+#async def process_ocel(file: UploadFile = File(...), parameters: Parameters = None):
+async def process_ocel(parameters: Parameters = Parameters()):
+    '''
+    # Save the uploaded file
+    file_location = f"/tmp/{file.filename}"
+    with open(file_location, "wb+") as file_object:
+        file_object.write(file.file.read())
+    '''
+    # Predefined filename
+    filename = "../ocsv/ocsv/EventLogs/BPI2017-Top10.jsonocel"
+    #parameters = {"execution_extraction": "leading_type",
+              #"leading_type": "application"}
+    # Load the OCEL file
+    ocel = ocel_import_factory.apply(file_path=filename, parameters=parameters)
+
+    # Step 1: Summarization Generation
+    all_summarizations, per_variant_dict, per_encoding_dict = IAVG.complete_intra_variant_summarization(ocel)
+
+    # Step 2: Summarization Matching
+    summarizations = SS.intra_variant_summarization_selection(all_summarizations, per_variant_dict, per_encoding_dict)
+
+    # Step 3: Inter-Variant Summarization
+    IEVG.NESTED_STRUCTURES = True
+    initial_set = [summarizations[i] for i in range(len(summarizations))]
+    hierarchies, final_super_variants = IEVG.generate_super_variant_hierarchy(
+        initial_set, parameters.max_levels, frequency_distribution_type=getattr(IEVG.Distribution, parameters.frequency_distribution_type)
+    )
+
+    # Extract super variants
+    values = []
+    for super_variant in final_super_variants[0]:
+        values.append("NEW SUPER VARIANT")
+        values.append(super_variant)
+
+    def extract_super_variants(nested_structure):
+        """Recursively extract SuperVariant objects from a nested structure."""
+        super_variants = []
+        if isinstance(nested_structure, tuple):
+            for item in nested_structure:
+                super_variants.extend(extract_super_variants(item))
+        elif isinstance(nested_structure, list):
+            for item in nested_structure:
+                super_variants.extend(extract_super_variants(item))
+        elif isinstance(nested_structure, SVD.SuperVariant):
+            super_variants.append(nested_structure)
+        return super_variants
+
+    # Create a useful data structure out of the nested list of super variants and tuples containing super variants.
+    super_variants_dict = {}
+    current_super_variant = None
+
+    for line in values:
+        if line == "NEW SUPER VARIANT":
+            current_super_variant = None
+        else:
+            if current_super_variant is None:
+                current_super_variant = []
+                super_variants_dict[len(super_variants_dict)] = current_super_variant
+            current_super_variant.extend(extract_super_variants(line))
+
+    def extract_hierarchy_info(hierarchy, level=0, info=None):
+        """Recursively extract information about the hierarchical structure."""
+        if info is None:
+            info = {"levels": {}, "super_variants": []}
+        
+        if isinstance(hierarchy, dict):
+            for key, value in hierarchy.items():
+                if key not in info["levels"]:
+                    info["levels"][key] = []
+                info["levels"][key].append(value)
+                extract_hierarchy_info(value, level + 1, info)
+        elif isinstance(hierarchy, list):
+            for item in hierarchy:
+                extract_hierarchy_info(item, level, info)
+        elif isinstance(hierarchy, tuple):
+            for item in hierarchy:
+                extract_hierarchy_info(item, level, info)
+        elif isinstance(hierarchy, SVD.SuperVariant):
+            info["super_variants"].append(hierarchy)
+        
+        return info
+
+    # Extract hierarchy information
+    hierarchy_info_list = []
+    for hierarchy in hierarchies:
+        hierarchy_info = extract_hierarchy_info(hierarchy)
+        hierarchy_info_list.append(hierarchy_info)
+
+    return {
+        "super_variants_dict": super_variants_dict,
+        "hierarchy_info_list": hierarchy_info_list
+    }
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+