Skip to content
Snippets Groups Projects
Commit b4c71243 authored by Simon Wolf's avatar Simon Wolf
Browse files

.gitignore prohibited parts of the commit. Try to fix it.

parent c75acc01
No related branches found
No related tags found
1 merge request!1Added loadability functionality to import OECL from Cortado and pass own data...
# Operating system files
.DS_Store
Thumbs.db
#backend
#src/backend/__pycache__/
#src/backend/api/
#src/backend/backend_utilities/
#src/backend/cache/
#src/backend/endpoints/
#src/backend/middleware/
#src/backend/core/
#frontend
# node_modules becomes very big after npm install
#src/frontend/node_modules
# after npm start
.angular/
\ No newline at end of file
No preview for this file type
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.get("/")
#@app.post("/process_ocel/")
#async def process_ocel(file: UploadFile = File(...), parameters: Parameters = None):
async def process_ocel(parameters: Parameters = Parameters(), filename: str = "/Users/swolf/Desktop/mining/ocsv/ocsv/EventLogs/BPI2017-Top10.jsonocel"):
'''
# 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
print("Processing OCEL file../n/n")
print(filename)
#filename = "../ocsv/ocsv/EventLogs/BPI2017-Top10.jsonocel"
#filename = "/Users/swolf/Desktop/mining/ocsv/ocsv/EventLogs/BPI2017-Top10.jsonocel"
#filename = "/Users/swolf/Desktop/mining/ocsv/ocsv/EventLogs/OCEL_example.jsonocel"
parameters = 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)
'''
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment