2 dependent workflows each composed of 2 independent cmdlinetool

Author: natthan-pigoux

pyproject.toml

@@ -44,6 +44,8 @@ pi-simulate-v2 = "dirac_cwl_proto.modules.pi_simulate_v2:app"
 pi-gather = "dirac_cwl_proto.modules.pi_gather:app"
 crypto = "dirac_cwl_proto.modules.crypto:app"
 lhcb-app = "dirac_cwl_proto.modules.lhcb_app:app"
+random-data-gen = "dirac_cwl_proto.modules.random_data_gen:app"
+gaussian-fit = "dirac_cwl_proto.modules.gaussian_fit:app"
 
 [tool.setuptools_scm]
 

src/dirac_cwl_proto/metadata_models.py

@@ -323,3 +323,32 @@ def post_process(self, job_path: Path):
         outputs = glob.glob(str(job_path / "mandelbrot_image*bmp"))
         if outputs:
             self._store_output("data-merged", outputs[0])
+
+
+class GaussianFitModel(IMetadataModel):
+    """Gaussian Fit metadata model."""
+
+    # Query parameters
+    data_file_1: str
+    data_file_2: str
+
+    # Input data
+    data: Path | None
+
+    def get_input_query(self, input_name: str) -> Path | None:
+        if input_name == "data_file_1":
+            return Path(self.data_file_1)
+        elif input_name == "data_file_2":
+            return Path(self.data_file_2)
+        return None
+
+    def get_output_query(self, output_name: str) -> Path | None:
+        if output_name == "fit-data" and self.data:
+            return self.data
+        return None
+
+    def post_process(self):
+        """Post process the outputs of a job."""
+        outputs = glob.glob("fit*.txt")
+        if outputs:
+            self._store_output("fit-data", outputs)

src/dirac_cwl_proto/modules/gaussian_fit.py

@@ -0,0 +1,29 @@
+import typer
+from rich.console import Console
+
+app = typer.Typer()
+console = Console()
+
+output: str = typer.Option("fit.txt", "--output", "-o", help="Output file")
+
+
+def gaussian_fit(input_data: str):
+    """Dummy gaussian fit."""
+    with open(input_data, "r") as f:
+        data = f.readlines()
+        mu = sum(map(float, data)) / len(data)
+        sigma = sum([(float(x) - mu) ** 2 for x in data]) / len(data)
+        console.print(f"Mean: {mu}, Std dev: {sigma}")
+        return mu, sigma
+
+
+@app.command()
+def main(input_data_files: list[str], output_data=output):
+    for input_data in input_data_files:
+        mu, sigma = gaussian_fit(input_data)
+        with open(output_data, "a") as f:
+            f.write(f"{input_data}: Mean: {mu}, Std dev: {sigma}\n")
+
+
+if __name__ == "__main__":
+    app()

src/dirac_cwl_proto/modules/random_data_gen.py

@@ -0,0 +1,22 @@
+import random
+
+import typer
+from rich.console import Console
+
+app = typer.Typer()
+console = Console()
+
+
+@app.command()
+def generate_random_data(file_path: str = "data.txt", num_lines: int = 100):
+    with open(file_path, "w") as f:
+        mu = random.randint(1, 10)
+        sig = random.randint(1, 5)
+        for _ in range(num_lines):
+            rd = random.gauss(mu, sig)
+            f.write(f"{rd}\n")
+    typer.echo(f"data file: {file_path}, mean: {mu}, std dev: {sig}")
+
+
+if __name__ == "__main__":
+    app()

test/test_workflows.py

@@ -71,7 +71,7 @@ def _cleanup():
         (
             "test/workflows/crypto/rot13.cwl",
             ["test/workflows/crypto/type_dependencies/job/inputs-crypto_complete.yaml"],
-        ),
+        ),  
         # Base64 only
         (
             "test/workflows/crypto/base64.cwl",
@@ -131,6 +131,28 @@ def _cleanup():
                 "test/workflows/mandelbrot/type_dependencies/job/inputs-mandelbrot_imagemerge.yaml"
             ],
         ),
+        # --- Gaussian fit example ---
+        # Complete
+        (
+            "test/workflows/gaussian_fit/main-workflow.cwl", 
+            [
+                "test/workflows/gaussian_fit/type_dependencies/production/metadata-gaussian-fit-complete.yaml"
+            ],
+        ),
+        # Data generation workflow
+        (
+            "test/workflows/gaussian_fit/data_generation/data-generation.cwl",
+            [
+                "test/workflows/gaussian_fit/type_dependencies/transformation/inputs-data-generation.yaml"
+            ],
+        ),
+        # Gaussian fit workflow
+        (
+            "test/workflows/gaussian_fit/gaussian_fit/gaussian-fit-workflow.cwl",
+            [
+                "test/workflows/gaussian_fit/type_dependencies/transformation/inputs-gaussian-fit.yaml"
+            ],
+        ),
     ],
 )
 def test_run_job_success(cli_runner, cleanup, cwl_file, inputs):

test/workflows/gaussian_fit/data_generation/data-generation-workflow.cwl

@@ -0,0 +1,38 @@
+cwlVersion: v1.2
+class: Workflow
+label: "Data Generation Workflow"
+doc: >
+  This workflow generates data using two independent data-generation tools.
+
+inputs:
+  data_file_name_1:
+    type: string
+  data_file_name_2:
+    type: string
+
+outputs:
+  data1:
+    type: File
+    outputSource: data-generation-1/data
+  data2:
+    type: File
+    outputSource: data-generation-2/data
+  log1:
+    type: File[]
+    outputSource: data-generation-1/log
+  log2:
+    type: File[]
+    outputSource: data-generation-2/log
+
+steps:
+  data-generation-1:
+    run: data-generation.cwl
+    in:
+      output_file_name: data_file_name_1
+    out: [data, log]
+
+  data-generation-2:
+    run: data-generation.cwl
+    in:
+      output_file_name: data_file_name_2
+    out: [data, log]

test/workflows/gaussian_fit/data_generation/data-generation.cwl

@@ -0,0 +1,22 @@
+cwlVersion: v1.2
+class: CommandLineTool
+label: "Benchmark Data Generation Tool"
+
+inputs:
+  output_file_name:
+    type: string
+    default: data.txt
+    inputBinding:
+      prefix: "--file-path"
+
+outputs:
+  data:
+    type: File
+    outputBinding:
+      glob: $(inputs.output_file_name)
+  log:
+    type: File[]?
+    outputBinding:
+      glob: "*.log"
+
+baseCommand: ["random-data-gen"]

test/workflows/gaussian_fit/gaussian_fit/gaussian-fit-workflow.cwl

@@ -0,0 +1,42 @@
+cwlVersion: v1.2
+class: Workflow
+label: "Fit Workflow"
+doc: >
+  This workflow runs Gaussian fit on the data generated by the data-generation workflow.
+
+requirements:
+  SubworkflowFeatureRequirement: {}
+  MultipleInputFeatureRequirement: {}
+
+inputs:
+  data1:
+    type: File
+  data2:
+    type: File
+
+outputs:
+  fit-data:
+    type: File[]
+    outputSource:
+      - fit-1/fit-data
+      - fit-2/fit-data
+    linkMerge: merge_flattened
+  log:
+    type: File[]
+    outputSource:
+      - fit-1/log
+      - fit-2/log
+    linkMerge: merge_flattened
+
+steps:
+  fit-1:
+    run: gaussian-fit.cwl
+    in:
+      data: data1
+    out: [fit-data, log]
+
+  fit-2:
+    run: gaussian-fit.cwl
+    in:
+      data: data2
+    out: [fit-data, log]

test/workflows/gaussian_fit/gaussian_fit/gaussian-fit.cwl

@@ -0,0 +1,21 @@
+cwlVersion: v1.2
+class: CommandLineTool
+label: "Gaussian Fit Tool"
+
+inputs:
+  data:
+    type: File
+    inputBinding:
+      position: 1
+
+outputs:
+  fit-data:
+    type: File[]
+    outputBinding:
+      glob: ["fit.txt"]
+  log:
+    type: File[]
+    outputBinding:
+      glob: ["fit.log"]
+
+baseCommand: ["gaussian-fit"]

test/workflows/gaussian_fit/main-workflow.cwl

@@ -0,0 +1,47 @@
+cwlVersion: v1.2
+class: Workflow
+label: "Main Workflow"
+doc: >
+  This workflow is composed of two dependent workflows
+  each composed of two command line tools:
+  - data-generation: produce data
+  - fit: run gaussian fit on data
+
+requirements:
+  SubworkflowFeatureRequirement: {}
+  MultipleInputFeatureRequirement: {}
+
+inputs:
+  data_file_name_1:
+    type: string
+    default: data_1.txt
+  data_file_name_2:
+    type: string
+    default: data_2.txt
+
+outputs:
+  fit-data:
+    type: File[]
+    outputSource:
+      - fit/fit-data
+    linkMerge: merge_flattened
+  logs:
+    type: File[]?
+    outputSource:
+      - fit/log
+    linkMerge: merge_flattened
+
+steps:
+  data-generation:
+    run: ./data_generation/data-generation-workflow.cwl
+    in:
+      data_file_name_1: data_file_name_1
+      data_file_name_2: data_file_name_2
+    out: [data1, data2, log1, log2]
+
+  fit:
+    run: ./gaussian_fit/gaussian-fit-workflow.cwl
+    in:
+      data1: data-generation/data1
+      data2: data-generation/data2
+    out: [fit-data, log]