<commit_before>"""Main entrypoint for cwl-tes."""
from __future__ import absolute_import, print_function, unicode_literals

import argparse
import os
import functools
import signal
import sys
import logging
import ftplib
import uuid
from typing import MutableMapping, MutableSequence
from typing_extensions import Text

import pkg_resources
from six.moves import urllib
from six import itervalues, StringIO

from ruamel import yaml
from schema_salad.sourceline import cmap
import cwltool.main
from cwltool.builder import substitute
from cwltool.process import scandeps, shortname
from cwltool.executors import MultithreadedJobExecutor, SingleJobExecutor
from cwltool.resolver import ga4gh_tool_registries
from cwltool.pathmapper import visit_class

from .tes import make_tes_tool, TESPathMapper
from .__init__ import __version__
from .ftp import FtpFsAccess

log = logging.getLogger("tes-backend")
log.setLevel(logging.INFO)
console = logging.StreamHandler()
# formatter = logging.Formatter("[%(asctime)s]\t[%(levelname)s]\t%(message)s")
# console.setFormatter(formatter)
log.addHandler(console)

DEFAULT_TMP_PREFIX = "tmp"


def versionstring():
    """Determine our version."""
    pkg = pkg_resources.require("cwltool")
    if pkg:
        cwltool_ver = pkg[0].version
    else:
        cwltool_ver = "unknown"
    return "%s %s with cwltool %s" % (sys.argv[0], __version__, cwltool_ver)


def ftp_upload(base_url, fs_access, cwl_obj):
    # type: (Text, FtpFsAccess, Dict[Text, Any]) -> None
    """
    Upload a File or Directory to the given FTP URL;

    Update the location URL to match.
    """
    if "path" not in cwl_obj and not (
            "location" in cwl_obj and cwl_obj["location"].startswith(
                "file:/")):
        return
    path = cwl_obj.get("path", cwl_obj["location"][6:])
    is_dir = os.path.isdir(path)
    basename = os.path.basename(path)
    dirname = os.path.dirname(path)
    if is_dir and cwl_obj["class"] != "Directory":
        raise ValueError("Passed a directory but Class is not Directory")
    if not is_dir and cwl_obj["class"] != "File":
        raise ValueError("Passed a file but Class is not File")
    try:
        fs_access.mkdir(base_url)
    except ftplib.all_errors:
        pass
    if not fs_access.isdir(base_url):
        raise Exception(
            'Failed to create target directory "{}".'.format(base_url))
    cwl_obj["location"] = base_url + '/' + basename
    cwl_obj.pop("path", None)
    if is_dir:
        if fs_access.isdir(fs_access.join(base_url, basename)):
            log.warning("FTP upload, Directory %s already exists", basename)
        else:
            for root, _subdirs, files in os.walk(path, followlinks=True):
                root_path = base_url + '/' + root[len(dirname):]
                fs_access.mkdir(root_path)
                for each_file in files:
                    with open(os.path.join(root,
                                           each_file), mode="rb") as source:
                        fs_access.upload(source, root_path + '/' + each_file)
        cwl_obj.pop("listing", None)
    else:
        if fs_access.isfile(fs_access.join(base_url, basename)):
            log.warning("FTP upload, file %s already exists", basename)
        else:
            with open(path, mode="rb") as source:
                fs_access.upload(source, cwl_obj["location"])


def main(args=None):
    """Main entrypoint for cwl-tes."""
    if args is None:
        args = sys.argv[1:]

    parser = arg_parser()
    parsed_args = parser.parse_args(args)

    if parsed_args.version:
        print(versionstring())
        return 0

    if parsed_args.tes is None:
        print(versionstring())
        parser.print_usage()
        print("cwl-tes: error: argument --tes is required")
        return 1

    if parsed_args.quiet:
        log.setLevel(logging.WARN)
    if parsed_args.debug:
        log.setLevel(logging.DEBUG)

    def signal_handler(*args):  # pylint: disable=unused-argument
        """setup signal handler"""
        log.info(
            "recieved control-c signal"
        )
        log.info(
            "terminating thread(s)..."
        )
        log.warning(
            "remote TES task(s) will keep running"
        )
        sys.exit(1)
    signal.signal(signal.SIGINT, signal_handler)

    ftp_cache = {}

    class CachingFtpFsAccess(FtpFsAccess):
        """Ensures that the FTP connection cache is shared."""
        def __init__(self, basedir):
            super(CachingFtpFsAccess, self).__init__(basedir, ftp_cache)
    ftp_fs_access = CachingFtpFsAccess(os.curdir)
    if parsed_args.remote_storage_url:
        parsed_args.remote_storage_url = ftp_fs_access.join(
            parsed_args.remote_storage_url, str(uuid.uuid4()))
    loading_context = cwltool.main.LoadingContext(vars(parsed_args))
    loading_context.construct_tool_object = functools.partial(
        make_tes_tool, url=parsed_args.tes,
        remote_storage_url=parsed_args.remote_storage_url)
    runtime_context = cwltool.main.RuntimeContext(vars(parsed_args))
    runtime_context.make_fs_access = CachingFtpFsAccess
    runtime_context.path_mapper = functools.partial(
        TESPathMapper, fs_access=ftp_fs_access)
    executor = functools.partial(
        tes_execute, job_executor=MultithreadedJobExecutor()
        if parsed_args.parallel else SingleJobExecutor(),
        loading_context=loading_context,
        remote_storage_url=parsed_args.remote_storage_url,
        ftp_access=ftp_fs_access)
    return cwltool.main.main(
        args=parsed_args,<commit_msg>unlimit the number of cores/memory<commit_after>"""Main entrypoint for cwl-tes."""
from __future__ import absolute_import, print_function, unicode_literals

import argparse
import os
import functools
import signal
import sys
import logging
import ftplib
import uuid
from typing import MutableMapping, MutableSequence
from typing_extensions import Text

import pkg_resources
from six.moves import urllib
from six import itervalues, StringIO

from ruamel import yaml
from schema_salad.sourceline import cmap
import cwltool.main
from cwltool.builder import substitute
from cwltool.process import scandeps, shortname
from cwltool.executors import MultithreadedJobExecutor, SingleJobExecutor
from cwltool.resolver import ga4gh_tool_registries
from cwltool.pathmapper import visit_class

from .tes import make_tes_tool, TESPathMapper
from .__init__ import __version__
from .ftp import FtpFsAccess

log = logging.getLogger("tes-backend")
log.setLevel(logging.INFO)
console = logging.StreamHandler()
# formatter = logging.Formatter("[%(asctime)s]\t[%(levelname)s]\t%(message)s")
# console.setFormatter(formatter)
log.addHandler(console)

DEFAULT_TMP_PREFIX = "tmp"


def versionstring():
    """Determine our version."""
    pkg = pkg_resources.require("cwltool")
    if pkg:
        cwltool_ver = pkg[0].version
    else:
        cwltool_ver = "unknown"
    return "%s %s with cwltool %s" % (sys.argv[0], __version__, cwltool_ver)


def ftp_upload(base_url, fs_access, cwl_obj):
    # type: (Text, FtpFsAccess, Dict[Text, Any]) -> None
    """
    Upload a File or Directory to the given FTP URL;

    Update the location URL to match.
    """
    if "path" not in cwl_obj and not (
            "location" in cwl_obj and cwl_obj["location"].startswith(
                "file:/")):
        return
    path = cwl_obj.get("path", cwl_obj["location"][6:])
    is_dir = os.path.isdir(path)
    basename = os.path.basename(path)
    dirname = os.path.dirname(path)
    if is_dir and cwl_obj["class"] != "Directory":
        raise ValueError("Passed a directory but Class is not Directory")
    if not is_dir and cwl_obj["class"] != "File":
        raise ValueError("Passed a file but Class is not File")
    try:
        fs_access.mkdir(base_url)
    except ftplib.all_errors:
        pass
    if not fs_access.isdir(base_url):
        raise Exception(
            'Failed to create target directory "{}".'.format(base_url))
    cwl_obj["location"] = base_url + '/' + basename
    cwl_obj.pop("path", None)
    if is_dir:
        if fs_access.isdir(fs_access.join(base_url, basename)):
            log.warning("FTP upload, Directory %s already exists", basename)
        else:
            for root, _subdirs, files in os.walk(path, followlinks=True):
                root_path = base_url + '/' + root[len(dirname):]
                fs_access.mkdir(root_path)
                for each_file in files:
                    with open(os.path.join(root,
                                           each_file), mode="rb") as source:
                        fs_access.upload(source, root_path + '/' + each_file)
        cwl_obj.pop("listing", None)
    else:
        if fs_access.isfile(fs_access.join(base_url, basename)):
            log.warning("FTP upload, file %s already exists", basename)
        else:
            with open(path, mode="rb") as source:
                fs_access.upload(source, cwl_obj["location"])


def main(args=None):
    """Main entrypoint for cwl-tes."""
    if args is None:
        args = sys.argv[1:]

    parser = arg_parser()
    parsed_args = parser.parse_args(args)

    if parsed_args.version:
        print(versionstring())
        return 0

    if parsed_args.tes is None:
        print(versionstring())
        parser.print_usage()
        print("cwl-tes: error: argument --tes is required")
        return 1

    if parsed_args.quiet:
        log.setLevel(logging.WARN)
    if parsed_args.debug:
        log.setLevel(logging.DEBUG)

    def signal_handler(*args):  # pylint: disable=unused-argument
        """setup signal handler"""
        log.info(
            "recieved control-c signal"
        )
        log.info(
            "terminating thread(s)..."
        )
        log.warning(
            "remote TES task(s) will keep running"
        )
        sys.exit(1)
    signal.signal(signal.SIGINT, signal_handler)

    ftp_cache = {}

    class CachingFtpFsAccess(FtpFsAccess):
        """Ensures that the FTP connection cache is shared."""
        def __init__(self, basedir):
            super(CachingFtpFsAccess, self).__init__(basedir, ftp_cache)
    ftp_fs_access = CachingFtpFsAccess(os.curdir)
    if parsed_args.remote_storage_url:
        parsed_args.remote_storage_url = ftp_fs_access.join(
            parsed_args.remote_storage_url, str(uuid.uuid4()))
    loading_context = cwltool.main.LoadingContext(vars(parsed_args))
    loading_context.construct_tool_object = functools.partial(
        make_tes_tool, url=parsed_args.tes,
        remote_storage_url=parsed_args.remote_storage_url)
    runtime_context = cwltool.main.RuntimeContext(vars(parsed_args))
    runtime_context.make_fs_access = CachingFtpFsAccess
    runtime_context.path_mapper = functools.partial(
        TESPathMapper, fs_access=ftp_fs_access)
    job_executor = MultithreadedJobExecutor() if parsed_args.parallel \
        else SingleJobExecutor()
    job_executor.max_ram = job_executor.max_cores = float("inf")
    executor = functools.partial(
        tes_execute, job_executor=job_executor,
        loading_context=loading_context,
        remote_storage_url=parsed_args.remote_storage_url,
        ftp_access=ftp_fs_access)
    return cwltool.main.main(
        args=parsed_args,
{'fraction_non_alphanumeric': 0.09953827028946902, 'fraction_numerical': 0.0015982951518380393, 'mean_word_length': 2.99680624556423, 'pattern_counts': {'":': 4, '<': 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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5674991', 'n_tokens_mistral': 3577, 'n_tokens_neox': 3482, 'n_words': 810}