from .nextflow_building_blocks import Nextflow_Building_Blocks from .bioflowinsighterror import BioFlowInsightError from .root import Root import re from .outils import * from . import constant class Main(Nextflow_Building_Blocks): def __init__(self, code, nextflow_file): Nextflow_Building_Blocks.__init__(self, code) self.nextflow_file = nextflow_file self.initialised = False self.root = None def get_calls_that_have_been_analysed(self): return self.nextflow_file.get_calls_that_have_been_analysed() def get_string_line(self, bit_of_code): return self.nextflow_file.get_string_line(bit_of_code) def get_modules_defined(self): return self.nextflow_file.get_modules_defined() def get_duplicate_status(self): return self.nextflow_file.get_duplicate_status() def get_DSL(self): return self.nextflow_file.get_DSL() def get_file_address(self): return self.nextflow_file.get_file_address() def get_output_dir(self): return self.nextflow_file.get_output_dir() def get_type(self): return "Main" def check_includes(self): code = self.get_code() pattern = constant.FULL_INCLUDE for match in re.finditer(pattern, code): if(self.get_type()=="Main"): raise BioFlowInsightError(f"An include ('{match.group(0)}') was found in the main in the file '{self.get_file_address()}'. FlowInsight does not support this -> see specification list.", num = 12,origin=self) elif(self.get_type()=="Subworkflow"): raise BioFlowInsightError(f"An include ('{match.group(0)}') was found in the subworkflow '{self.get_name()}' in the file '{self.get_file_address()}'. FlowInsight does not support this -> see specification list.", num = 12, origin=self) else: raise Exception("This shouldn't happen!") def initialise(self): if(not self.initialised): self.initialised=True #Get the modules (Processes defined for the main/subworkflow) self.modules_defined = self.nextflow_file.get_modules_defined() #Check that includes are not defined in the main or subworkflows self.check_includes() self.root = Root(code=self.get_code(), origin=self, modules_defined=self.modules_defined) self.root.initialise() def get_structure(self, dico): self.root.get_structure(dico) return dico