<commit_before>import os
import re
import glob
import ruamel.yaml
import yaml
import logging
import shutil
from natsort import natsorted

from tmlib.workflow.jterator.utils import get_module_directories
from tmlib.readers import YamlReader
from tmlib.writers import YamlWriter
from tmlib.errors import PipelineOSError

logger = logging.getLogger(__name__)

HANDLES_SUFFIX = '.handles.yaml'
PIPE_SUFFIX = '.pipe.yaml'


def list_projects(directory):
    '''Lists Jterator projects in a given directory.
    A Jterator project is defined as a folder containing a `.pipe` file.

    Parameters
    ----------
    directory: str
        absolute path to a directory
    '''
    projects = [
        os.path.join(directory, name)
        for name in os.listdir(directory)
        if os.path.isdir(os.path.join(directory, name)) and
        glob.glob(os.path.join(directory, name, '*%s' % PIPE_SUFFIX))
    ]
    if not projects:
        logger.warning('No Jterator projects found in %s' % directory)
    return projects


class Project(object):

    '''A Jterator project is defined as a folder containing a `.pipe` file.
    The class holds information about the project, in particular on the content
    of YAML pipeline and module descriptor files that can be edited in the
    JtUI app.
    '''
    def __init__(self, step_location, pipe_name, pipe=None, handles=None):
        '''
        Parameters
        ----------
        step_location: str
            path to the project folder
        pipe_name: str
            name of the pipeline
        pipe: dict, optional
            pipeline description (default: ``None``)
        handles: List[dict], optional
            module descriptions (default: ``None``)
        '''
        self.step_location = step_location
        self.pipe_name = pipe_name
        self.pipe = pipe
        self.handles = handles

    @property
    def pipe(self):
        '''dict: name and description of the pipeline'''
        if self._pipe is None:
            self._pipe = self._create_pipe()
        return self._pipe

    @pipe.setter
    def pipe(self, value):
        self._pipe = value

    @property
    def handles(self):
        '''List[dict]: name and description of modules'''
        if self._handles is None:
            self._handles = self._create_handles()
        return self._handles

    @handles.setter
    def handles(self, value):
        self._handles = value

    def _get_pipe_file(self, directory=None):
        if not directory:
            directory = self.step_location
        pipe_files = glob.glob(
            os.path.join(directory, '%s%s' % (self.pipe_name, PIPE_SUFFIX))
        )
        if len(pipe_files) == 1:
            return pipe_files[0]
        elif len(pipe_files) > 1:
            raise PipelineOSError(
                'More than more pipeline descriptor file found: %s' % directory
            )
        else:
            raise PipelineOSError(
                'No pipeline descriptor file found: %s' % directory
            )

    def _get_handles_files(self, directory=None):
        if not directory:
            directory = os.path.join(self.step_location, 'handles')
        else:
            directory = os.path.join(directory, 'handles')
        handles_files = glob.glob(
            os.path.join(directory, '*%s' % HANDLES_SUFFIX)
        )
        if not handles_files:
            # We don't raise an exception, because an empty handles folder
            # can occur, for example upon creation of a new project
            logger.warning('No handles files found: %s' % directory)
        return handles_files

    @staticmethod
    def _get_descriptor_name(filename):
        return os.path.splitext(
            os.path.splitext(os.path.basename(filename))[0]
        )[0]

    @staticmethod
    def _replace_values(old, new):
        # Recursively replace values in the `old` mapping with those of `new`.
        # NOTE: This is not a general approach, but targeted for the design of
        # the pipeline and module descriptor files.
        # TODO: make this general using a recursive strategy
        for k1, v1 in old.iteritems():
            # 'project', 'images', and 'description' keys in the pipeline
            # descriptor file or 'input' and 'output' keys in the module
            # descriptor file
            if isinstance(v1, dict):
                # this affects 'project' and 'jobs'
                for k2, v2 in v1.iteritems():
                    if isinstance(v2, list):
                        # this is the array of 'layers' in the 'images' section
                        old[k1][k2] = []
                        for i, element in enumerate(new[k1][k2]):
                            if isinstance(element, dict):
                                old[k1][k2].append(new[k1][k2][i])

                    else:
                        old[k1][k2] = new[k1][k2]
            elif isinstance(v1, list):
                # this affects the 'description' key in the pipeline descriptor
                # file and the 'input' and 'output' keys in the module
                # descriptor file
                old[k1] = []
                for i, element in enumerate(new[k1]):
                    if isinstance(element, dict):
                        old[k1].append(new[k1][i])
            else:
                old[k1] = new[k1]
        return old

    def _create_pipe(self):
        pipe_file = self._get_pipe_file()
        with YamlReader(pipe_file) as f:
            pipe = {
                'name': self._get_descriptor_name(pipe_file),
                'description': f.read()
            }
        if pipe['description']['pipeline']:
            # Add module 'name' to pipeline for display in the interface
            for i, module in enumerate(pipe['description']['pipeline']):
                pipe['description']['pipeline'][i]['name'] = \
                    self._get_descriptor_name(
                        pipe['description']['pipeline'][i]['handles']
                    )
        return pipe

    @property
    def _module_names(self):
        return [m['name'] for m in self.pipe['description']['pipeline']]

    def _create_handles(self):
        handles = list()
        handles_files = self._get_handles_files()
        if handles_files:
            for h_file in handles_files:
                if not os.path.isabs(h_file):
                    h_file = os.path.join(self.step_location, h_file)
                if not os.path.exists(h_file):
                    raise PipelineOSError(
                        'Handles file does not exist: "%s"' % h_file
                    )
                with YamlReader(h_file) as f:
                    handles.append({
                        'name': self._get_descriptor_name(h_file),
                        'description': f.read()
                    })
        # Sort handles information according to order of modules in the pipeline
        names = [h['name'] for h in handles]
        sorted_handles = list()
        for name in self._module_names:
            if name not in names:
                raise ValueError(
                    'Handles for module "%s" does not exist.' % name
                )
            sorted_handles.append(handles[names.index(name)])

        return sorted_handles

    def _create_pipe_file(self, repo_dir):
        pipe_file = os.path.join(
            self.step_location,
            '%s%s' % (self.pipe_name, PIPE_SUFFIX)
        )
        pipe_skeleton = {
            'project': {
                'description': str()
            },
            'images': {
                'planes': [
                    {'name': str(), 'correct': True}
                ]
            },
            'pipeline': list()
        }
        with YamlWriter(pipe_file) as f:
            f.write(pipe_skeleton, use_ruamel=True)
<commit_msg>Check pipeline upon construction of pipe object<commit_after>import os
import re
import glob
import ruamel.yaml
import yaml
import logging
import shutil
from natsort import natsorted

from tmlib.workflow.jterator.utils import get_module_directories
from tmlib.readers import YamlReader
from tmlib.writers import YamlWriter
from tmlib.errors import PipelineOSError
from tmlib.errors import PipelineDescriptionError

logger = logging.getLogger(__name__)

HANDLES_SUFFIX = '.handles.yaml'
PIPE_SUFFIX = '.pipe.yaml'


def list_projects(directory):
    '''Lists Jterator projects in a given directory.
    A Jterator project is defined as a folder containing a `.pipe` file.

    Parameters
    ----------
    directory: str
        absolute path to a directory
    '''
    projects = [
        os.path.join(directory, name)
        for name in os.listdir(directory)
        if os.path.isdir(os.path.join(directory, name)) and
        glob.glob(os.path.join(directory, name, '*%s' % PIPE_SUFFIX))
    ]
    if not projects:
        logger.warning('No Jterator projects found in %s' % directory)
    return projects


class Project(object):

    '''A Jterator project is defined as a folder containing a `.pipe` file.
    The class holds information about the project, in particular on the content
    of YAML pipeline and module descriptor files that can be edited in the
    JtUI app.
    '''
    def __init__(self, step_location, pipe_name, pipe=None, handles=None):
        '''
        Parameters
        ----------
        step_location: str
            path to the project folder
        pipe_name: str
            name of the pipeline
        pipe: dict, optional
            pipeline description (default: ``None``)
        handles: List[dict], optional
            module descriptions (default: ``None``)
        '''
        self.step_location = step_location
        self.pipe_name = pipe_name
        self.pipe = pipe
        self.handles = handles

    @property
    def pipe(self):
        '''dict: name and description of the pipeline'''
        if self._pipe is None:
            self._pipe = self._create_pipe()
        return self._pipe

    @pipe.setter
    def pipe(self, value):
        self._pipe = value

    @property
    def handles(self):
        '''List[dict]: name and description of modules'''
        if self._handles is None:
            self._handles = self._create_handles()
        return self._handles

    @handles.setter
    def handles(self, value):
        self._handles = value

    def _get_pipe_file(self, directory=None):
        if not directory:
            directory = self.step_location
        pipe_files = glob.glob(
            os.path.join(directory, '%s%s' % (self.pipe_name, PIPE_SUFFIX))
        )
        if len(pipe_files) == 1:
            return pipe_files[0]
        elif len(pipe_files) > 1:
            raise PipelineOSError(
                'More than more pipeline descriptor file found: %s' % directory
            )
        else:
            raise PipelineOSError(
                'No pipeline descriptor file found: %s' % directory
            )

    def _get_handles_files(self, directory=None):
        if not directory:
            directory = os.path.join(self.step_location, 'handles')
        else:
            directory = os.path.join(directory, 'handles')
        handles_files = glob.glob(
            os.path.join(directory, '*%s' % HANDLES_SUFFIX)
        )
        if not handles_files:
            # We don't raise an exception, because an empty handles folder
            # can occur, for example upon creation of a new project
            logger.warning('No handles files found: %s' % directory)
        return handles_files

    @staticmethod
    def _get_descriptor_name(filename):
        return os.path.splitext(
            os.path.splitext(os.path.basename(filename))[0]
        )[0]

    @staticmethod
    def _replace_values(old, new):
        # Recursively replace values in the `old` mapping with those of `new`.
        # NOTE: This is not a general approach, but targeted for the design of
        # the pipeline and module descriptor files.
        # TODO: make this general using a recursive strategy
        for k1, v1 in old.iteritems():
            # 'project', 'images', and 'description' keys in the pipeline
            # descriptor file or 'input' and 'output' keys in the module
            # descriptor file
            if isinstance(v1, dict):
                # this affects 'project' and 'jobs'
                for k2, v2 in v1.iteritems():
                    if isinstance(v2, list):
                        # this is the array of 'layers' in the 'images' section
                        old[k1][k2] = []
                        for i, element in enumerate(new[k1][k2]):
                            if isinstance(element, dict):
                                old[k1][k2].append(new[k1][k2][i])

                    else:
                        old[k1][k2] = new[k1][k2]
            elif isinstance(v1, list):
                # this affects the 'description' key in the pipeline descriptor
                # file and the 'input' and 'output' keys in the module
                # descriptor file
                old[k1] = []
                for i, element in enumerate(new[k1]):
                    if isinstance(element, dict):
                        old[k1].append(new[k1][i])
            else:
                old[k1] = new[k1]
        return old

    def _create_pipe(self):
        pipe_file = self._get_pipe_file()
        with YamlReader(pipe_file) as f:
            pipe = {
                'name': self._get_descriptor_name(pipe_file),
                'description': f.read()
            }
        if 'pipeline' not in pipe['description']:
            raise PipelineDescriptionError(
                'Pipeline descriptor file "%s" must contain key "pipeline".'
                % pipe_file
            )
        if pipe['description']['pipeline']:
            if not isinstance(pipe['description']['pipeline'], list):
                raise PipelineDescriptionError(
                    'Pipeline description in "%s" must be an array.'
                    % pipe_file
                )
            # Add module 'name' to pipeline for display in the interface
            for i, module in enumerate(pipe['description']['pipeline']):
                if 'handles' not in pipe['description']['pipeline'][i]:
                    raise PipelineDescriptionError(
                        'Element #%d of "pipeline" array in pipeline '
                        'descriptor file "%s" must contain key "handles".'
                        % (i, pipe_file)
                    )
                pipe['description']['pipeline'][i]['name'] = \
                    self._get_descriptor_name(
                        pipe['description']['pipeline'][i]['handles']
                    )
        else:
            logger.warn('no pipeline description provided in "%s"', pipe_file)
        return pipe

    @property
    def _module_names(self):
        return [m['name'] for m in self.pipe['description']['pipeline']]

    def _create_handles(self):
        handles = list()
        handles_files = self._get_handles_files()
        if handles_files:
            for h_file in handles_files:
                if not os.path.isabs(h_file):
                    h_file = os.path.join(self.step_location, h_file)
                if not os.path.exists(h_file):
                    raise PipelineOSError(
                        'Handles file does not exist: "%s"' % h_file
                    )
                with YamlReader(h_file) as f:
                    handles.append({
                        'name': self._get_descriptor_name(h_file),
                        'description': f.read()
                    })
        # Sort handles information according to order of modules in the pipeline
        names = [h['name'] for h in handles]
        sorted_handles = list()
        for name in self._module_names:
            if name not in names:
                raise ValueError(
                    'Handles for module "%s" does not exist.' % name
                )
            sorted_handles.append(handles[names.index(name)])

        return sorted_handles

    def _create_pipe_file(self, repo_dir):
        pipe_file = os.path.join(
            self.step_location,
            '%s%s' % (self.pipe_name, PIPE_SUFFIX)
        )
        pipe_skeleton = {
            'project': {
                'description': str()
            },
            'input': {
                'channels': [
                    {'name': str(), 'correct': True}
                ]
            },
            'pipeline': list()
        }
        with YamlWriter(pipe_file) as f:
            f.write(pipe_skeleton, use_ruamel=True)

{'fraction_non_alphanumeric': 0.09497744360902256, 'fraction_numerical': 0.003729323308270677, 'mean_word_length': 1.7376914210439651, 'pattern_counts': {'":': 0, '<': 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': '3001685', 'n_tokens_mistral': 4541, 'n_tokens_neox': 4396, 'n_words': 1314}