Nextflow with Flowcept

Author: rafaelfsilva

wfcommons/wfbench/translator/templates/nextflow_templates/flowcept.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2025 The WfCommons Team.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+import ast
+import logging
+import pathlib
+import sys
+import time
+from flowcept.flowcept_api.flowcept_controller import Flowcept
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[WfBench][%%(asctime)s][%%(levelname)s] %%(message)s",
+    datefmt="%%H:%%M:%%S",
+    handlers=[logging.StreamHandler()]
+)
+
+workflow_name = sys.argv[1]
+workflow_id = sys.argv[2]
+out_files = ast.literal_eval(sys.argv[3])
+
+logging.info("Flowcept Starting")
+flowcept_agent = Flowcept(workflow_id=workflow_id, workflow_name=workflow_name, bundle_exec_id=workflow_id, start_persistence=False, save_workflow=True)
+
+try:
+    flowcept_agent.start()
+except Exception:
+    import traceback
+    traceback.print_exc()
+
+remaining_files = set(out_files)
+
+while remaining_files:
+    found_files = set()
+    for f in remaining_files:
+        if pathlib.Path(f).exists():
+            found_files.add(f)
+    remaining_files -= found_files
+    if not remaining_files:
+        break
+    time.sleep(1)
+    
+try:
+    flowcept_agent.stop()
+except Exception:
+    import traceback
+    traceback.print_exc()
+
+logging.info("Flowcept Completed")

wfcommons/wfbench/translator/templates/nextflow_templates/workflow.nf

@@ -0,0 +1,42 @@
+params.simulate = false
+params.pwd = null
+params.help = null
+pwd = null
+
+def printUsage(error_msg, exit_code) {
+    def usage_string = """
+Usage: nextflow run workflow.nf --pwd /path/to/directory [--simulate] [--help]
+
+    Required parameters:
+      --pwd         Working directory (where the workflow.nf file is located)
+
+    Optional parameters:
+      --help        Show this message and exit.
+      --simulate    Use a "sleep 1" for all tasks instead of the WfBench benchmark.
+"""
+    if (error_msg) {
+        def RED = '\u001B[31m'
+        def RESET = '\u001B[0m'
+        System.err.println "${RED}Error: ${RESET}" + error_msg
+    }
+    System.err.println usage_string
+    exit exit_code
+}
+
+def validateParams() {
+    if (params.help) {
+        printUsage(msg = "", exit_code=0)
+    }
+    if (params.pwd == null) {
+        printUsage(msg = "Missing required parameter: --pwd", exit_code=1)
+    }
+    pwd = file(params.pwd).toAbsolutePath().toString()
+    if (!file(pwd).exists()) {
+        printUsage(msg = "Directory not found: ${pwd}", exit_code=1)
+    } 
+}
+
+// Call validation at the start
+validateParams()
+
+# Generated code goes here