from uuid import UUID from api.od import ODAPI from examples.ftg_pm_pt.ftg_pm_pt import FtgPmPt from examples.ftg_pm_pt.runner import FtgPmPtRunner def find_previous_artefact(od: ODAPI, linked_artefacts): return next((od.get_source(link) for link in linked_artefacts if not od.get_incoming(od.get_source(link), "pt_PrevVersion")), None) def create_activity_links(od: ODAPI, activity, prev_element, ctrl_port, end_trace=None, relation_type="pt_IsFollowedBy"): od.create_link(None, "pt_RelatesTo", activity, ctrl_port) od.create_link(None, relation_type, prev_element, activity) if end_trace: od.create_link(None, "pt_IsFollowedBy", activity, end_trace) def get_workflow_path(od: ODAPI, activity: UUID): return od.get_slot_value(activity, "subworkflow_path") def get_workflow(workflow_path: str): with open(workflow_path, "r") as f: return f.read() ############################ def get_runtime_state(od: ODAPI, design_obj: UUID): states = od.get_incoming(design_obj, "pm_Of") if len(states) == 0: print(f"Design object '{od.get_name(design_obj)}' has no runtime state.") return None return od.get_source(states[0]) def get_source_incoming(od: ODAPI, obj: UUID, link_name: str): links = od.get_incoming(obj, link_name) if len(links) == 0: print(f"Object '{od.get_name(obj)} has no incoming links of type '{link_name}'.") return None return od.get_source(links[0]) def get_target_outgoing(od: ODAPI, obj: UUID, link_name: str): links = od.get_outgoing(obj, link_name) if len(links) == 0: print(f"Object '{od.get_name(obj)} has no outgoing links of type '{link_name}'.") return None return od.get_target(links[0]) def set_control_port_value(od: ODAPI, port: UUID, value: bool): state = get_runtime_state(od, port) od.set_slot_value(state, "active", value) def set_artefact_data(od: ODAPI, artefact: UUID, value: bytes): state = artefact # Only the proces model of the artefact contains a runtime state if od.get_type_name(state) == "pm_Artefact": state = get_runtime_state(od, artefact) od.set_slot_value(state, "data", value) def get_artefact_data(od: ODAPI, artefact): state = artefact # Only the proces model of the artefact contains a runtime state if od.get_type_name(state) == "pm_Artefact": state = get_runtime_state(od, artefact) return od.get_slot_value(state, "data") ############################ def set_workflow_control_source(workflow_model: FtgPmPt, ctrl_port_name: str, composite_linkage: dict): od = workflow_model.odapi source_port_name = composite_linkage[ctrl_port_name] source_port = od.get(source_port_name) set_control_port_value(od, source_port, True) def set_workflow_artefacts(act_od: ODAPI, activity: UUID, workflow_model: FtgPmPt, composite_linkage: dict): for data_port in [act_od.get_target(data_in) for data_in in act_od.get_outgoing(activity, "pm_HasDataIn")]: # Get the data source port of the inner workflow data_port_name = act_od.get_name(data_port) source_port_name = composite_linkage[data_port_name] source_port = workflow_model.odapi.get(source_port_name) # Get the artefact that is linked to the data port of the activity act_artefact = get_source_incoming(act_od, data_port, "pm_DataFlowOut") # Get the data of the artefact artefact_data = get_artefact_data(act_od, act_artefact) # Get the artefact that is linked to the data port of the inner workflow workflow_artefact = get_target_outgoing(workflow_model.odapi, source_port, "pm_DataFlowIn") set_artefact_data(workflow_model.odapi, workflow_artefact, artefact_data) def get_activity_port_from_inner_port(composite_linkage: dict, port_name: str): for act_port_name, work_port_name in composite_linkage.items(): if work_port_name == port_name: return act_port_name def execute_composite_workflow(od: ODAPI, activity: UUID, ctrl_port: UUID, composite_linkage: dict, packages: dict | None, matched=None): activity_name = od.get_slot_value(activity, "name") # First get the path of the object diagram file that contains the inner workflow of the activity workflow_path = get_workflow_path(od, activity) # Read the object diagram file workflow = get_workflow(workflow_path) # Create an FtgPmPt object workflow_model = FtgPmPt(activity_name) # Load the workflow to the object workflow_model.load_model(workflow) # Set the correct control source port of the workflow to active set_workflow_control_source(workflow_model, od.get_name(ctrl_port), composite_linkage[activity_name]) # If a data port is linked, set the data of the artefact set_workflow_artefacts(od, activity, workflow_model, composite_linkage[activity_name]) # Create an FtgPmPtRunner object with the FtgPmPt object workflow_runner = FtgPmPtRunner(workflow_model) # Set the packages if present workflow_runner.set_packages(packages, is_path=False) # Run the FtgPmPtRunner (is a subprocess necessary? This makes it more complicated because now we have direct access to the object) workflow_runner.run() # Contains all the ports of the inner workflow -> map back to the activity ports, and so we can set the correct # Control ports to active and also set the data artefacts correctly ports = extract_inner_workflow(workflow_model.odapi) start_act = None end_act = None for port in [port for port in ports if port]: port_name = workflow_model.odapi.get_name(port) activity_port_name = get_activity_port_from_inner_port(composite_linkage[activity_name], port_name) activity_port = od.get(activity_port_name) match workflow_model.odapi.get_type_name(port): case "pm_CtrlSource": start_act = handle_control_source(od, activity_port, matched("prev_trace_element")) case "pm_CtrlSink": end_act = handle_control_sink(od, activity_port, start_act, matched("end_trace")) case "pm_DataSource": handle_data_source(od, activity_port, start_act) case "pm_DataSink": handle_data_sink(od, workflow_model.odapi, activity_port, port, end_act) def handle_control_source(od: ODAPI, port, prev_trace_elem): set_control_port_value(od, port, False) start_activity = od.create_object(None, "pt_StartActivity") create_activity_links(od, start_activity, prev_trace_elem, port) return start_activity def handle_control_sink(od: ODAPI, port, start_act, end_trace): set_control_port_value(od, port, True) end_activity = od.create_object(None, "pt_EndActivity") create_activity_links(od, end_activity, start_act, port, end_trace) return end_activity def handle_data_source(od: ODAPI, port, start_activity): pt_artefact = od.create_object(None, "pt_Artefact") od.create_link(None, "pt_Consumes", pt_artefact, start_activity) pm_artefact = get_source_incoming(od, port, "pm_DataFlowOut") pm_artefact_data = get_artefact_data(od, pm_artefact) set_artefact_data(od, pt_artefact, pm_artefact_data) prev_pt_artefact = find_previous_artefact(od, od.get_incoming(pm_artefact, "pt_BelongsTo")) if prev_pt_artefact: od.create_link(None, "pt_PrevVersion", pt_artefact, prev_pt_artefact) od.create_link(None, "pt_BelongsTo", pt_artefact, pm_artefact) def handle_data_sink(act_od: ODAPI, work_od: ODAPI, act_port, work_port, end_activity): pt_artefact = act_od.create_object(None, "pt_Artefact") act_od.create_link(None, "pt_Produces", end_activity, pt_artefact) work_artefact = get_source_incoming(work_od, work_port, "pm_DataFlowOut") work_artefact_data = get_artefact_data(work_od, work_artefact) act_artefact = get_target_outgoing(act_od, act_port, "pm_DataFlowIn") set_artefact_data(act_od, act_artefact, work_artefact_data) set_artefact_data(act_od, pt_artefact, work_artefact_data) prev_pt_artefact = find_previous_artefact(act_od, act_od.get_incoming(act_artefact, "pt_BelongsTo")) if prev_pt_artefact: act_od.create_link(None, "pt_PrevVersion", pt_artefact, prev_pt_artefact) act_od.create_link(None, "pt_BelongsTo", pt_artefact, act_artefact) def extract_inner_workflow(workflow: ODAPI): # Get the model, this should be only one name, model = workflow.get_all_instances("pm_Model")[0] # Get the start of the process trace start_trace = get_source_incoming(workflow, model, "pt_Starts") # Get the end of the process trace end_trace = get_source_incoming(workflow, model, "pt_Ends") # Get the first started activity first_activity = get_target_outgoing(workflow, start_trace, "pt_IsFollowedBy") # Get the last ended activity end_activity = get_source_incoming(workflow, end_trace, "pt_IsFollowedBy") # Get the control port that started the activity act_ctrl_in = get_target_outgoing(workflow, first_activity, "pt_RelatesTo") # Get the control port that is activated when the activity is executed act_ctrl_out = get_target_outgoing(workflow, end_activity, "pt_RelatesTo") # Get the control source of the workflow ports = [] for port in workflow.get_incoming(act_ctrl_in, "pm_CtrlFlow"): source = workflow.get_source(port) if workflow.get_type_name(source) == "pm_CtrlSource": # Only one port can activate an activity ports.append(source) break # Get the control sink of the workflow for port in workflow.get_outgoing(act_ctrl_out, "pm_CtrlFlow"): sink = workflow.get_target(port) if workflow.get_type_name(sink) == "pm_CtrlSink": # Only one port can be set to active one an activity is ended ports.append(sink) break # Get the data port that the activity consumes (if used) consumed_links = workflow.get_incoming(first_activity, "pt_Consumes") if len(consumed_links) > 0: pt_artefact = None for link in consumed_links: pt_artefact = workflow.get_source(link) # Check if it is the first artefact -> contains no previous version if len(workflow.get_outgoing(pt_artefact, "pt_PrevVersion")) == 0: break pm_artefact = get_target_outgoing(workflow, pt_artefact, "pt_BelongsTo") # Find the data source port for link in workflow.get_incoming(pm_artefact, "pm_DataFlowIn"): source = workflow.get_source(link) if workflow.get_type_name(source) == "pm_DataSource": # An activity can only use one artefact as input ports.append(source) break # Get all data ports that are connected to an artefact that is produced by an activity in the workflow, # where the artefact is also part of main workflow for port_name, data_sink in workflow.get_all_instances("pm_DataSink"): pm_art = get_source_incoming(workflow, data_sink, "pm_DataFlowOut") # If the pm_artefact is linked to a proces trace artefact that is produced, we can add to port links = workflow.get_incoming(pm_art, "pt_BelongsTo") if not len(links): continue # A data sink port linkage will only be added to the proces trace when an activity is ended and so an artefact # is produced, meaning that if a belongsTo link exists, a proces trace artefact is linked to this data port ports.append(data_sink) return ports