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"
Parses the .nextflow.log file for signatures of pipeline status.
        It sets the :attr:`status_info` attribute.