finish skeletons for all envs

src/utils/operandi_utils/oton/cli.py

@@ -13,17 +13,21 @@ def cli():
               show_default=True, help='Path to the OCR-D workflow file to be converted.')
 @click.option('-O', '--output_path', type=click.Path(dir_okay=False, writable=True),
               show_default=True, help='Path of the Nextflow workflow script to be generated.')
-@click.option('-D', '--dockerized', is_flag=True,
-              help='If set, then the dockerized variant of the Nextflow script is generated.')
-def convert(input_path: str, output_path: str, dockerized: bool):
+@click.option('-E', '--environment', type=str, default="local",
+              help='The environment of the output Nextflow file. One of: local, docker, apptainer.')
+def convert(input_path: str, output_path: str, environment: str):
     print(f"Converting from: {input_path}")
     print(f"Converting to: {output_path}")
-    if dockerized:
+    if environment == "local":
+        OTONConverter().convert_oton_env_local(input_path, output_path)
+    elif environment == "docker":
         OTONConverter().convert_oton_env_docker(input_path, output_path)
-        print("Success: Converting workflow from ocrd process to Nextflow with docker processor calls")
+    elif environment == "apptainer":
+        OTONConverter().convert_oton_env_apptainer(input_path, output_path)
     else:
-        OTONConverter().convert_oton_env_local(input_path, output_path)
-        print("Success: Converting workflow from ocrd process to Nextflow with local processor calls")
+        print("Unspecified environment type. Must be one of: local, docker, apptainer.")
+        exit(1)
+    print(f"Success: Converting workflow from ocrd process to Nextflow with {environment} processor calls")
 
 
 @cli.command("validate", help="Validate an OCR-D workflow txt file.")

src/utils/operandi_utils/oton/constants.py

@@ -25,7 +25,8 @@
     "REPR_INPUT_FILE_GRP",
     "REPR_METS_PATH",
     "REPR_WORKSPACE_DIR",
-    "SPACES"
+    "SPACES",
+    "WORKFLOW_COMMENT"
 ]
 
 OCRD_ALL_JSON_FILE = resource_filename(__name__, 'ocrd_all_tool.json')

src/utils/operandi_utils/oton/nf_block_workflow.py

@@ -10,28 +10,27 @@ def __init__(self, workflow_name: str, nf_processes: List[NextflowBlockProcess])
         self.logger.setLevel(getLevelName(OTON_LOG_LEVEL))
 
         self.workflow_name = workflow_name
-        self.nf_blocks_process: List[NextflowBlockProcess] = nf_processes
-
-    def file_representation(self):
-        representation = 'workflow {\n'
-        representation += f'{SPACES}{self.workflow_name}:\n'
-
-        self.logger.info(self.nf_blocks_process)
+        self.workflow_calls: List[str] = []
+        self.produce_workflow_calls(nf_processes)
 
+    def produce_workflow_calls(self, nf_blocks_process: List[NextflowBlockProcess]):
         previous_nfp = None
-        for nfp in self.nf_blocks_process:
-            nfp_1 = nfp.processor_call_arguments.input_file_grps
-            nfp_2 = nfp.processor_call_arguments.output_file_grps
-            representation += f'{SPACES}{SPACES}{nfp.nf_process_name}('
+        for block_process in nf_blocks_process:
+            in_file_grps = block_process.processor_call_arguments.input_file_grps
+            out_file_grps = block_process.processor_call_arguments.output_file_grps
+            workflow_call = f"{block_process.nf_process_name}("
             if previous_nfp is None:
-                representation += f'{PARAMS_KEY_METS_PATH}, {PARAMS_KEY_INPUT_FILE_GRP}, "{nfp_2}")'
+                workflow_call += f'{PARAMS_KEY_METS_PATH}, {PARAMS_KEY_INPUT_FILE_GRP}, "{out_file_grps}"'
             else:
-                representation += f'{previous_nfp}.out, "{nfp_1}", "{nfp_2}")'
-            representation += f'\n'
-            previous_nfp = nfp.nf_process_name
+                workflow_call += f'{previous_nfp}.out, "{in_file_grps}", "{out_file_grps}"'
+            workflow_call += ")\n"
+            previous_nfp = block_process.nf_process_name
+            self.workflow_calls.append(workflow_call)
 
+    def file_representation(self):
+        representation = 'workflow {\n'
+        representation += f'{SPACES}{self.workflow_name}:\n'
+        for workflow_call in self.workflow_calls:
+            representation += f"{SPACES}{SPACES}{workflow_call}"
         representation += '}'
-
-        self.logger.debug(f"\n{representation}")
-        self.logger.info(f"Successfully created Nextflow Workflow: {self.workflow_name}")
         return representation

src/utils/operandi_utils/oton/nf_file_executable.py

@@ -1,5 +1,5 @@
 from logging import getLevelName, getLogger
-from typing import List, Tuple
+from typing import List
 
 from operandi_utils.oton.ocrd_validator import ProcessorCallArguments
 from operandi_utils.oton.constants import (
@@ -42,13 +42,20 @@ def build_parameters_docker(self):
         self.nf_lines_parameters.append(REPR_INPUT_FILE_GRP)
         self.nf_lines_parameters.append(REPR_METS_PATH)
         self.nf_lines_parameters.append(REPR_WORKSPACE_DIR)
-
         self.nf_lines_parameters.append(REPR_ENV_WRAPPER)
 
         self.nf_lines_parameters.append('')
 
     def build_parameters_apptainer(self):
-        raise NotImplemented("This feature is not implemented yet!")
+        self.nf_lines_parameters.append('nextflow.enable.dsl = 2')
+        self.nf_lines_parameters.append('')
+
+        self.nf_lines_parameters.append(REPR_INPUT_FILE_GRP)
+        self.nf_lines_parameters.append(REPR_METS_PATH)
+        self.nf_lines_parameters.append(REPR_WORKSPACE_DIR)
+        self.nf_lines_parameters.append(REPR_ENV_WRAPPER)
+
+        self.nf_lines_parameters.append('')
 
     def build_nextflow_processes_local(self, ocrd_processor: List[ProcessorCallArguments]):
         index = 0
@@ -60,7 +67,6 @@ def build_nextflow_processes_local(self, ocrd_processor: List[ProcessorCallArgum
             nf_process_block.add_parameter_input(parameter=DIR_OUT, parameter_type='val')
             nf_process_block.add_parameter_output(parameter=METS_FILE, parameter_type='path')
             self.nf_blocks_process.append(nf_process_block)
-            self.logger.info(f"Successfully created Nextflow Process: {nf_process_block.nf_process_name}")
             index += 1
 
     def build_nextflow_processes_docker(self, ocrd_processor: List[ProcessorCallArguments]):
@@ -73,11 +79,19 @@ def build_nextflow_processes_docker(self, ocrd_processor: List[ProcessorCallArgu
             nf_process_block.add_parameter_input(parameter=DIR_OUT, parameter_type='val')
             nf_process_block.add_parameter_output(parameter=METS_FILE, parameter_type='path')
             self.nf_blocks_process.append(nf_process_block)
-            self.logger.info(f"Successfully created Nextflow Process: {nf_process_block.nf_process_name}")
             index += 1
 
-    def build_nextflow_processes_apptainer(self, ocrd_processor: List[ProcessorCallArguments]) -> Tuple[List[str], str]:
-        raise NotImplemented("This feature is not implemented yet!")
+    def build_nextflow_processes_apptainer(self, ocrd_processor: List[ProcessorCallArguments]):
+        index = 0
+        for processor in ocrd_processor:
+            nf_process_block = NextflowBlockProcess(processor, index, env_wrapper=True)
+            nf_process_block.add_directive(directive='maxForks', value='1')
+            nf_process_block.add_parameter_input(parameter=METS_FILE, parameter_type='path')
+            nf_process_block.add_parameter_input(parameter=DIR_IN, parameter_type='val')
+            nf_process_block.add_parameter_input(parameter=DIR_OUT, parameter_type='val')
+            nf_process_block.add_parameter_output(parameter=METS_FILE, parameter_type='path')
+            self.nf_blocks_process.append(nf_process_block)
+            index += 1
 
     def __assign_first_file_grps_param(self):
         first_file_grps = self.nf_blocks_process[0].processor_call_arguments.input_file_grps

src/utils/operandi_utils/oton/oton_converter.py

@@ -6,25 +6,28 @@ class OTONConverter:
     def __init__(self):
         self.ocrd_validator = OCRDValidator()
 
-    def convert_oton_env_local(self, input_path: str, output_path: str) -> NextflowFileExecutable:
+    def __convert_oton(self, input_path: str, output_path: str, environment: str):
         self.ocrd_validator.validate(input_path)
         list_processor_call_arguments = self.ocrd_validator.validate(input_path)
         nf_file_executable = NextflowFileExecutable()
-        nf_file_executable.build_parameters_local()
-        nf_file_executable.build_nextflow_processes_local(ocrd_processor=list_processor_call_arguments)
+        if environment == "local":
+            nf_file_executable.build_parameters_local()
+            nf_file_executable.build_nextflow_processes_local(ocrd_processor=list_processor_call_arguments)
+        elif environment == "docker":
+            nf_file_executable.build_parameters_docker()
+            nf_file_executable.build_nextflow_processes_docker(ocrd_processor=list_processor_call_arguments)
+        elif environment == "apptainer":
+            nf_file_executable.build_parameters_apptainer()
+            nf_file_executable.build_nextflow_processes_apptainer(ocrd_processor=list_processor_call_arguments)
         nf_file_executable.build_main_workflow()
         nf_file_executable.produce_nextflow_file(output_path)
         return nf_file_executable
 
+    def convert_oton_env_local(self, input_path: str, output_path: str) -> NextflowFileExecutable:
+        return self.__convert_oton(input_path, output_path, environment="local")
+
     def convert_oton_env_docker(self, input_path: str, output_path: str) -> NextflowFileExecutable:
-        self.ocrd_validator.validate(input_path)
-        list_processor_call_arguments = self.ocrd_validator.validate(input_path)
-        nf_file_executable = NextflowFileExecutable()
-        nf_file_executable.build_parameters_docker()
-        nf_file_executable.build_nextflow_processes_docker(ocrd_processor=list_processor_call_arguments)
-        nf_file_executable.build_main_workflow()
-        nf_file_executable.produce_nextflow_file(output_path)
-        return nf_file_executable
+        return self.__convert_oton(input_path, output_path, environment="docker")
 
     def convert_oton_env_apptainer(self, input_path: str, output_path: str) -> NextflowFileExecutable:
-        raise NotImplemented("This feature is not implemented yet!")
+        return self.__convert_oton(input_path, output_path, environment="apptainer")