flowcraft/generator/inspect.py
def _update_pipeline_status(self):
        """Parses the .nextflow.log file for signatures of pipeline status.
        It sets the :attr:`status_info` attribute.
        """

        with open(self.log_file) as fh:

            try:
                first_line = next(fh)
            except:
                raise eh.InspectionError("Could not read .nextflow.log file. Is file empty?")
            time_str = " ".join(first_line.split()[:2])
            self.time_start = time_str

            if not self.execution_command:
                try:
                    self.execution_command = re.match(
                        ".*nextflow run (.*)", first_line).group(1)
                except AttributeError:
                    self.execution_command = "Unknown"

            for line in fh:

                if "DEBUG nextflow.cli.CmdRun" in line:
                    if not self.nextflow_version:
                        try:
                            vline = next(fh)
                            self.nextflow_version = re.match(
                                ".*Version: (.*)", vline).group(1)
                        except AttributeError:
                            self.nextflow_version = "Unknown"

                if "Session aborted" in line:
                    self.run_status = "aborted"
                    # Get abort cause
                    try:
                        self.abort_cause = re.match(
                            ".*Cause: (.*)", line).group(1)
                    except AttributeError:
                        self.abort_cause = "Unknown"
                    # Get time of pipeline stop
                    time_str = " ".join(line.split()[:2])
                    self.time_stop = time_str
                    self.send = True
                    return
                if "Execution complete -- Goodbye" in line:
                    self.run_status = "complete"
                    # Get time of pipeline stop
                    time_str = " ".join(line.split()[:2])
                    self.time_stop = time_str
                    self.send = True
                    return

        if self.run_status not in ["running", ""]:
            self._clear_inspect()
            # Take a break to allow nextflow to restart before refreshing
            # pipeine processes
            sleep(5)
            self._get_pipeline_processes()

        self.run_status = "running"
def _update_pipeline_status(self):
        """Parses the .nextflow.log file for signatures of pipeline status.
        It sets the :attr:`status_info` attribute.
        """

        with open(self.log_file) as fh:

            try:
                first_line = next(fh)
            except:
                raise eh.InspectionError("Could not read .nextflow.log file. Is file empty?")
            time_str = " ".join(first_line.split()[:2])
            self.time_start = time_str

            if not self.execution_command:
                try:
                    self.execution_command = re.match(
                        ".*nextflow run (.*)", first_line).group(1)
                except AttributeError:
                    self.execution_command = "Unknown"

            for line in fh:

                if "DEBUG nextflow.cli.CmdRun" in line:
                    if not self.nextflow_version:
                        try:
                            vline = next(fh)
                            self.nextflow_version = re.match(
                                ".*Version: (.*)", vline).group(1)
                        except AttributeError:
                            self.nextflow_version = "Unknown"

                if "Session aborted" in line:
                    self.run_status = "aborted"
                    # Get abort cause
                    try:
                        self.abort_cause = re.match(
                            ".*Cause: (.*)", line).group(1)
                    except AttributeError:
                        self.abort_cause = "Unknown"
                    # Get time of pipeline stop
                    time_str = " ".join(line.split()[:2])
                    self.time_stop = time_str
                    self.send = True
                    return
                if "Execution complete -- Goodbye" in line:
                    self.run_status = "complete"
                    # Get time of pipeline stop
                    time_str = " ".join(line.split()[:2])
                    self.time_stop = time_str
                    self.send = True
                    return

        if self.run_status not in ["running", ""]:
            self._clear_inspect()
            # Take a break to allow nextflow to restart before refreshing
            # pipeine processes
            sleep(5)
            self._get_pipeline_processes()

        self.run_status = "running"
def _update_pipeline_status ( self ) : with open ( self . log_file ) as fh : try : first_line = next ( fh ) except : raise eh . InspectionError ( "Could not read .nextflow.log file. Is file empty?" ) time_str = " " . join ( first_line . split ( ) [ : 2 ] ) self . time_start = time_str if not self . execution_command : try : self . execution_command = re . match ( ".*nextflow run (.*)" , first_line ) . group ( 1 ) except AttributeError : self . execution_command = "Unknown" for line in fh : if "DEBUG nextflow.cli.CmdRun" in line : if not self . nextflow_version : try : vline = next ( fh ) self . nextflow_version = re . match ( ".*Version: (.*)" , vline ) . group ( 1 ) except AttributeError : self . nextflow_version = "Unknown" if "Session aborted" in line : self . run_status = "aborted" # Get abort cause try : self . abort_cause = re . match ( ".*Cause: (.*)" , line ) . group ( 1 ) except AttributeError : self . abort_cause = "Unknown" # Get time of pipeline stop time_str = " " . join ( line . split ( ) [ : 2 ] ) self . time_stop = time_str self . send = True return if "Execution complete -- Goodbye" in line : self . run_status = "complete" # Get time of pipeline stop time_str = " " . join ( line . split ( ) [ : 2 ] ) self . time_stop = time_str self . send = True return if self . run_status not in [ "running" , "" ] : self . _clear_inspect ( ) # Take a break to allow nextflow to restart before refreshing # pipeine processes sleep ( 5 ) self . _get_pipeline_processes ( ) self . run_status = "running"
Parses the .nextflow.log file for signatures of pipeline status.
        It sets the :attr:`status_info` attribute.
Parses the . nextflow . log file for signatures of pipeline status . It sets the : attr : status_info attribute .