<commit_before>"""Module to generate and work with process flowcharts"""
import json
import logging
import threading
import time

import requests

from bcbio_monitor import parser as ps
from dateutil.parser import parse
from graphviz import Digraph
from paramiko import client

logger = logging.getLogger(__name__)

class RunData(Digraph):
    """Container for the information of a particular run within an analsysis"""

    def __init__(self, id):
        # Initializes hte Digraph structure
        super(RunData, self).__init__(comment='bcbio flow chart', format='png', encoding='UTF8')

        self.ID = id
        self._nodes = []
        self.steps = []
        self.errored = False


    def json(self):
        """Retrun JSON representation for this run"""
        return {
            "id": self.ID,
            "steps": self.steps,
            "graph_source": self.source,
            "errored": self.errored
        }


class AnalysisData(object):
    """Representation for a bcbio-nextgen analysis log"""

    def __init__(self, logfile, host='localhost', port='5000', update=True, remote=None):
        """Initialices a AnalysisData object.

        :param logfile: str - Path to the logfile where to extract information
        :param host: str - Host address where the monitor is running
        :param port: str - Port where the monitor is listening
        :param update: boolean - Update frontend on every line read
        :param remote: dict - Connection parameters if the log is on a remote host.
        """
        self.logfile = logfile
        self.update = update
        self.remote = remote
        self.base_url = "http://{}".format(':'.join([host, port]))
        self.FIRST_STEP = None
        self.analysis_finished = False
        self.finished_reading = False
        self.runs = [RunData(1)]
        self.current_run = 0
        self._last_message = {'line': ''}
        self._reading_thread = threading.Thread(target=self.follow_log)
        # Daemonise the thread so that it's killed with Ctrl+C
        self._reading_thread.daemon = True
        self._reading_thread.start()


    def new_run(self):
        """Creates a new RunData object and increments pointers"""
        self.current_run += 1
        self.runs.append(RunData(self.current_run + 1))


    def follow_log(self):
        """Reads a logfile continuously and updates internal graph if new step is found"""
        # Server needs to be up and running before starting sending POST requests
        time.sleep(5)
        try:
            if self.remote:
                logger.debug('Logfile in remote host!')
                cl = client.SSHClient()
                # Try to load system keys
                cl.load_system_host_keys()
                cl.connect(self.remote['host'], port=self.remote.get('port', 22), username=self.remote.get('username', None), \
                           password=self.remote.get('password', None))
                sftp = cl.open_sftp()
                f = sftp.open(self.logfile, 'r')
            else:
                f = open(self.logfile, 'r')
        except IOError:
            raise RuntimeError("Provided logfile does not exist or its not readable")
        self.analysis_finished = False
        last_line_read = False
        while not self.analysis_finished:
            line = f.readline()
            if not line:
                self.finished_reading = True
                if not last_line_read:
                    self.update_frontend({'finished_reading': True})
                    if self.update:
                        self.update_frontend(self._last_message)
                last_line_read = True
                time.sleep(1)
                continue
            parsed_line = ps.parse_log_line(line)
            self._last_message = parsed_line
            self.analysis_finished = parsed_line['step'] == 'finished'

            # If this is a new step, update internal data
            parsed_line['new_run'] = False
            if parsed_line['step'] and not parsed_line['step'] == 'error':
                if self.FIRST_STEP is None:
                    self.FIRST_STEP = parsed_line['step']
                elif parsed_line['step'] == self.FIRST_STEP:
                    parsed_line['new_run'] = True
                    self.new_run()
                node_id = 'run-{}_'.format(self.current_run + 1) + '_'.join(parsed_line['step'].lower().split())
                parsed_line['step_id'] = node_id
                self.runs[self.current_run].steps.append(parsed_line)
                self.runs[self.current_run].node(node_id, parsed_line['step'])
                self.runs[self.current_run]._nodes.append(node_id)
                n_nodes = len(self.runs[self.current_run]._nodes)
                if n_nodes > 1:
                    self.runs[self.current_run].edge(self.runs[self.current_run]._nodes[n_nodes - 2], self.runs[self.current_run]._nodes[n_nodes -1])
                parsed_line['graph_source'] = self.runs[self.current_run].source

            elif parsed_line['step'] == 'error':
                self.runs[self.current_run].errored = True

            # Update frontend only if its a new step _or_ the update flag is set to true and we are
            # not loading the log for the first time
            if (last_line_read and self.update) or parsed_line['step']:
                self.update_frontend(parsed_line)
        f.close()<commit_msg>Handle SSH disconnections (resolves #25)<commit_after>"""Module to generate and work with process flowcharts"""
import json
import logging
import os
import threading
import time

import requests

from bcbio_monitor import parser as ps
from dateutil.parser import parse
from graphviz import Digraph
from paramiko import client
from socket import timeout, error

logger = logging.getLogger(__name__)

class RunData(Digraph):
    """Container for the information of a particular run within an analsysis"""

    def __init__(self, id):
        # Initializes hte Digraph structure
        super(RunData, self).__init__(comment='bcbio flow chart', format='png', encoding='UTF8')

        self.ID = id
        self._nodes = []
        self.steps = []
        self.errored = False


    def json(self):
        """Retrun JSON representation for this run"""
        return {
            "id": self.ID,
            "steps": self.steps,
            "graph_source": self.source,
            "errored": self.errored
        }


class AnalysisData(object):
    """Representation for a bcbio-nextgen analysis log"""

    def __init__(self, logfile, host='localhost', port='5000', update=True, remote=None):
        """Initialices a AnalysisData object.

        :param logfile: str - Path to the logfile where to extract information
        :param host: str - Host address where the monitor is running
        :param port: str - Port where the monitor is listening
        :param update: boolean - Update frontend on every line read
        :param remote: dict - Connection parameters if the log is on a remote host.
        """
        self.logfile = logfile
        self.update = update
        self.remote = remote
        self.base_url = "http://{}".format(':'.join([host, port]))
        self.FIRST_STEP = None
        self.analysis_finished = False
        self.finished_reading = False
        self.runs = [RunData(1)]
        self.current_run = 0
        self._last_message = {'line': ''}
        self._reading_thread = threading.Thread(target=self.follow_log)
        # Daemonise the thread so that it's killed with Ctrl+C
        self._reading_thread.daemon = True
        self._reading_thread.start()


    def new_run(self):
        """Creates a new RunData object and increments pointers"""
        self.current_run += 1
        self.runs.append(RunData(self.current_run + 1))


    def follow_log(self):
        """Reads a logfile continuously and updates internal graph if new step is found"""
        # Server needs to be up and running before starting sending POST requests
        time.sleep(5)
        try:
            if self.remote:
                logger.debug('Logfile in remote host!')
                cl = client.SSHClient()
                # Try to load system keys
                cl.load_system_host_keys()
                cl.connect(self.remote['host'], port=self.remote.get('port', 22), username=self.remote.get('username', None), \
                           password=self.remote.get('password', None))
                sftp = cl.open_sftp()
                f = sftp.open(self.logfile, 'r')
                f.settimeout(5) # Set 5 seconds timeout for read operations
            else:
                f = open(self.logfile, 'r')
        except IOError:
            raise RuntimeError("Provided logfile does not exist or its not readable")
        self.analysis_finished = False
        last_line_read = False
        while not self.analysis_finished:
            try:
                line = f.readline()
            except timeout:
                logger.error("Connection with the server lost, trying to reconnect and continue reading")
                current_pos = f.tell()
                try:
                    cl.connect(self.remote['host'], port=self.remote.get('port', 22), username=self.remote.get('username', None), \
                               password=self.remote.get('password', None), timeout=300, banner_timeout=300)
                except error:
                    logger.error("Couldn't connect to the server after 5 minutes, aborting.")
                    os._exit(0)
                else:
                    logger.info("Connection restablished!! Will continue reading the logfile")
                sftp = cl.open_sftp()
                f = sftp.open(self.logfile, 'r')
                f.seek(current_pos)
            else:
                if not line:
                    self.finished_reading = True
                    if not last_line_read:
                        self.update_frontend({'finished_reading': True})
                        if self.update:
                            self.update_frontend(self._last_message)
                    last_line_read = True
                    time.sleep(1)
                    continue
                parsed_line = ps.parse_log_line(line)
                self._last_message = parsed_line
                self.analysis_finished = parsed_line['step'] == 'finished'

                # If this is a new step, update internal data
                parsed_line['new_run'] = False
                if parsed_line['step'] and not parsed_line['step'] == 'error':
                    if self.FIRST_STEP is None:
                        self.FIRST_STEP = parsed_line['step']
                    elif parsed_line['step'] == self.FIRST_STEP:
                        parsed_line['new_run'] = True
                        self.new_run()
                    node_id = 'run-{}_'.format(self.current_run + 1) + '_'.join(parsed_line['step'].lower().split())
                    parsed_line['step_id'] = node_id
                    self.runs[self.current_run].steps.append(parsed_line)
                    self.runs[self.current_run].node(node_id, parsed_line['step'])
                    self.runs[self.current_run]._nodes.append(node_id)
                    n_nodes = len(self.runs[self.current_run]._nodes)
                    if n_nodes > 1:
                        self.runs[self.current_run].edge(self.runs[self.current_run]._nodes[n_nodes - 2], self.runs[self.current_run]._nodes[n_nodes -1])
                    parsed_line['graph_source'] = self.runs[self.current_run].source

                elif parsed_line['step'] == 'error':
                    self.runs[self.current_run].errored = True

                # Update frontend only if its a new step _or_ the update flag is set to true and we are
                # not loading the log for the first time
                if (last_line_read and self.update) or parsed_line['step']:
                    self.update_frontend(parsed_line)
        f.close()
{'fraction_non_alphanumeric': 0.09806698883519413, 'fraction_numerical': 0.003999333444425929, 'mean_word_length': 1.9527675276752767, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2744115', 'n_tokens_mistral': 3350, 'n_tokens_neox': 3159, 'n_words': 873}