<commit_before>##########################################################################
#
#   MRC FGU Computational Genomics Group
#
#   $Id$
#
#   Copyright (C) 2016 <NAME>
#
#   This program is free software; you can redistribute it and/or
#   modify it under the terms of the GNU General Public License
#   as published by the Free Software Foundation; either version 2
#   of the License, or (at your option) any later version.
#
#   This program is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU General Public License for more details.
#
#   You should have received a copy of the GNU General Public License
#   along with this program; if not, write to the Free Software
#   Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
##########################################################################
'''Cluster.py - cluster utility functions for ruffus pipelines
==============================================================

This module abstracts the DRMAA native specification and provides
convenience functions for running Drmaa jobs.


Reference
---------

'''

import re
import os
import stat

try:
    import drmaa
    HAS_DRMAA = True
except RuntimeError:
    HAS_DRMAA = False

def setupDrmaaJobTemplate(drmaa_session, options, job_name, job_memory):
    '''Sets up a Drmma job template. Currently SGE and SLURM are
       supported'''

    if not job_memory:
        raise ValueError("Job memory must be specified when running"
                         "DRMAA jobs")

    jt = drmaa_session.createJobTemplate()
    jt.workingDirectory = options["workingdir"]
    jt.jobEnvironment = {'BASH_ENV': '~/.bashrc'}
    jt.args = []
    if not re.match("[a-zA-Z]", job_name[0]):
        job_name = "_" + job_name

    # queue manager specific configuration options
    queue_manager = options["cluster_queue_manager"]

    if queue_manager.lower() == "sge":

        # see: ? cannot find documentation on the SGE native spec

        spec = ["-V",
                "-N %s" % job_name]

        if options["cluster_priority"]:
            spec.append("-p %(cluster_priority)i")

        if options["cluster_options"]:
            spec.append("%(cluster_options)s")

        if not options["cluster_memory_resource"]:
            raise ValueError("The cluster memory resource must be specified")

        for resource in options["cluster_memory_resource"].split(","):
            spec.append("-l %s=%s" % (resource, job_memory))

        # if process has multiple threads, use a parallel environment
        if 'job_threads' in options:
            spec.append(
                "-pe %(cluster_parallel_environment)s %(job_threads)i -R y")

            if "cluster_pe_queue" in options and 'job_threads' in options:
                spec.append(
                    "-q %(cluster_pe_queue)s")
        else:
            spec.append("-q %(cluster_queue)s")

    elif queue_manager.lower() == "slurm":

        # SLURM DOCS:
        # http://apps.man.poznan.pl/trac/slurm-drmaa
        # https://computing.llnl.gov/linux/slurm/cons_res_share.html
        #
        # The SLURM Consumable Resource plugin is required
        # The "CR_CPU_Memory" resource must be specified
        #
        # i.e. in slurm.conf:
        # SelectType=select/cons_res
        # SelectTypeParameters=CR_CPU_Memory
        #
        # * Note that --cpus-per-task will actually refer to cores
        #   with the appropriate Node configuration
        #
        # SLURM-DRMAA DOCS - Note that version 1.2 (SVN) is required
        # http://apps.man.poznan.pl/trac/slurm-drmaa
        #
        # Not implemented:
        # -V: SLURM automatically passess the environment variables
        # -p: does not appear to be part of the slurm drmaa native spec
        #
        # TODO: add "--account" (not sure the best way to fill param).

        spec = ["-J %s" % job_name]

        if options["cluster_options"]:
            spec.append("%(cluster_options)s")

        if 'job_threads' in options:
            job_threads = options["job_threads"]
        else:
            job_threads = 1  # probably should come from a config option

        spec.append("--cpus-per-task=%s" % job_threads)

        # Note the that the specified memory must be per CPU
        # for consistency with the implemented SGE approach

        if job_memory.endswith("G"):
            job_memory_per_cpu = int(job_memory[:-1]) * 1000
        elif job_memory.endswith("M"):
            job_memory_per_cpu = int(job_memory[:-1])
        else:
            raise ValueError('job memory unit not recognised for SLURM, '
                             'must be either "M" (for Mb) or "G" (for Gb),'
                             ' e.g. 1G or 1000M for 1 Gigabyte of memory')

        spec.append("--mem-per-cpu=%s" % job_memory_per_cpu)

    else:
        raise ValueError("Queue manager %s not supported" % queue_manager)

    jt.nativeSpecification = " ".join(spec) % options

    # keep stdout and stderr separate
    jt.joinFiles = False

    return jt


def setDrmaaJobPaths(job_template, job_path):<commit_msg>Add partition (queue) selection to SLURM native spec<commit_after>##########################################################################
#
#   MRC FGU Computational Genomics Group
#
#   $Id$
#
#   Copyright (C) 2016 <NAME>
#
#   This program is free software; you can redistribute it and/or
#   modify it under the terms of the GNU General Public License
#   as published by the Free Software Foundation; either version 2
#   of the License, or (at your option) any later version.
#
#   This program is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU General Public License for more details.
#
#   You should have received a copy of the GNU General Public License
#   along with this program; if not, write to the Free Software
#   Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
##########################################################################
'''Cluster.py - cluster utility functions for ruffus pipelines
==============================================================

This module abstracts the DRMAA native specification and provides
convenience functions for running Drmaa jobs.


Reference
---------

'''

import re
import os
import stat

try:
    import drmaa
    HAS_DRMAA = True
except RuntimeError:
    HAS_DRMAA = False

def setupDrmaaJobTemplate(drmaa_session, options, job_name, job_memory):
    '''Sets up a Drmma job template. Currently SGE and SLURM are
       supported'''

    if not job_memory:
        raise ValueError("Job memory must be specified when running"
                         "DRMAA jobs")

    jt = drmaa_session.createJobTemplate()
    jt.workingDirectory = options["workingdir"]
    jt.jobEnvironment = {'BASH_ENV': '~/.bashrc'}
    jt.args = []
    if not re.match("[a-zA-Z]", job_name[0]):
        job_name = "_" + job_name

    # queue manager specific configuration options
    queue_manager = options["cluster_queue_manager"]

    if queue_manager.lower() == "sge":

        # see: ? cannot find documentation on the SGE native spec

        spec = ["-V",
                "-N %s" % job_name]

        if options["cluster_priority"]:
            spec.append("-p %(cluster_priority)i")

        if options["cluster_options"]:
            spec.append("%(cluster_options)s")

        if not options["cluster_memory_resource"]:
            raise ValueError("The cluster memory resource must be specified")

        for resource in options["cluster_memory_resource"].split(","):
            spec.append("-l %s=%s" % (resource, job_memory))

        # if process has multiple threads, use a parallel environment
        if 'job_threads' in options:
            spec.append(
                "-pe %(cluster_parallel_environment)s %(job_threads)i -R y")

            if "cluster_pe_queue" in options and 'job_threads' in options:
                spec.append(
                    "-q %(cluster_pe_queue)s")
        else:
            spec.append("-q %(cluster_queue)s")

    elif queue_manager.lower() == "slurm":

        # SLURM DOCS:
        # http://apps.man.poznan.pl/trac/slurm-drmaa
        # https://computing.llnl.gov/linux/slurm/cons_res_share.html
        #
        # The SLURM Consumable Resource plugin is required
        # The "CR_CPU_Memory" resource must be specified
        #
        # i.e. in slurm.conf:
        # SelectType=select/cons_res
        # SelectTypeParameters=CR_CPU_Memory
        #
        # * Note that --cpus-per-task will actually refer to cores
        #   with the appropriate Node configuration
        #
        # SLURM-DRMAA DOCS - Note that version 1.2 (SVN) is required
        # http://apps.man.poznan.pl/trac/slurm-drmaa
        #
        # Not implemented:
        # -V: SLURM automatically passess the environment variables
        # -p: does not appear to be part of the slurm drmaa native spec
        #
        # TODO: add "--account" (not sure the best way to fill param).

        spec = ["-J %s" % job_name]

        if options["cluster_options"]:
            spec.append("%(cluster_options)s")

        if 'job_threads' in options:
            job_threads = options["job_threads"]
        else:
            job_threads = 1  # probably should come from a config option

        spec.append("--cpus-per-task=%s" % job_threads)

        # Note the that the specified memory must be per CPU
        # for consistency with the implemented SGE approach

        if job_memory.endswith("G"):
            job_memory_per_cpu = int(job_memory[:-1]) * 1000
        elif job_memory.endswith("M"):
            job_memory_per_cpu = int(job_memory[:-1])
        else:
            raise ValueError('job memory unit not recognised for SLURM, '
                             'must be either "M" (for Mb) or "G" (for Gb),'
                             ' e.g. 1G or 1000M for 1 Gigabyte of memory')

        spec.append("--mem-per-cpu=%s" % job_memory_per_cpu)

        # set the partition to use (equivalent of SGE queue)
        spec.append("--partition=%(cluster_queue)s")

    else:
        raise ValueError("Queue manager %s not supported" % queue_manager)

    jt.nativeSpecification = " ".join(spec) % options

    # keep stdout and stderr separate
    jt.joinFiles = False

    return jt


def setDrmaaJobPaths(job_template, job_path):
{'fraction_non_alphanumeric': 0.13234186746987953, 'fraction_numerical': 0.006588855421686747, 'mean_word_length': 2.5240464344941955, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, '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': '2993367', 'n_tokens_mistral': 3079, 'n_tokens_neox': 2942, 'n_words': 966}