<commit_before>import os
import logging
import requests
import ftplib
import concurrent.futures as cf
from parsl.executors.base import ParslExecutor
from parsl.data_provider.globus import get_globus
from parsl.app.app import App

logger = logging.getLogger(__name__)


class DataManager(ParslExecutor):
    """DataManager uses the familiar Executor interface, where staging tasks are submitted
    to it, and DataFutures are returned.
    """

    """
    In general a site where remote file is going to be stage in is unknown until
    Executer submits an app that depends on the file. However, in most practical
    cases, a site where an app is executed and a file needs to be staged in is
    known. Such cases should be detected by DataManager to optimize file
    transfers. Possible cases are:
    1. Config defines one site only.
    2. Config defines several sites but all of the sites share the filesystem /
       use the same Globus endpoint.
    3. Config defines several sites with different Globus endpoints but a user
       specified explicitely that apps must be executed on a particular site.
    """

    default_data_manager = None

    @classmethod
    def get_data_manager(cls, dfk=None, config=None):
        if not cls.default_data_manager or dfk:
            cls.default_data_manager = DataManager(dfk, config=config)
        return cls.default_data_manager

    def __init__(self, dfk, max_workers=10, config=None):
        """Initialize the DataManager.

        Kwargs:
           - max_workers (int) : Number of threads (Default=10)
           - config (dict): The config dict object for the site:

        """
        self._scaling_enabled = False
        self.executor = cf.ThreadPoolExecutor(max_workers=max_workers)

        self.dfk = dfk
        self.config = config
        if not self.config:
            self.config = {"sites": []}
        self.files = []
        self.globus = None

    def submit(self, *args, **kwargs):
        """Submit a staging app. All optimization should be here."""
        return self.executor.submit(*args, **kwargs)

    def scale_in(self, blocks, *args, **kwargs):
        pass

    def scale_out(self, *args, **kwargs):
        pass

    def shutdown(self, block=False):
        """Shutdown the ThreadPool.

        Kwargs:
            - block (Bool): To block for confirmations or not

        """
        x = self.executor.shutdown(wait=block)
        logger.debug("Done with executor shutdown")
        return x

    def scaling_enabled(self):
        return self._scaling_enabled

    def add_file(self, file):
        if file.scheme == 'globus':
            if not self.globus:
                self.globus = get_globus()
            # keep a list of all remote files for optimization purposes (TODO)
            self.files.append(file)
            self._set_local_path(file)

    def _set_local_path(self, file):
        globus_ep = self._get_globus_site()
        file.local_path = os.path.join(
                globus_ep['working_dir'],
                file.filename)

    def _get_globus_site(self, site_name='all'):
        if isinstance(site_name, list):
            site_name = site_name[0]
        for s in self.config['sites']:
            if site_name is 'all' or s['site'] == site_name:
                if 'data' not in s:
                    continue
                data = s['data']
                if 'globus' not in data or 'working_dir' not in data:
                    continue
                globus_ep = data['globus']
                if 'endpoint_name' not in globus_ep:
                    continue
                endpoint_name = data['globus']['endpoint_name']
                working_dir = os.path.normpath(data['working_dir'])
                if 'endpoint_path' in globus_ep and 'local_path' in globus_ep:
                    endpoint_path = os.path.normpath(globus_ep['endpoint_path'])
                    local_path = os.path.normpath(globus_ep['local_path'])
                    common_path = os.path.commonpath((local_path, working_dir))
                    if local_path != common_path:
                        raise Exception('"local_path" must be equal or an absolute subpath of "working_dir"')
                    relative_path = os.path.relpath(working_dir, common_path)
                    endpoint_path = os.path.join(endpoint_path, relative_path)
                else:
                    endpoint_path = working_dir
                return {'site_name': s['site'],
                        'endpoint_name': endpoint_name,
                        'endpoint_path': endpoint_path,
                        'working_dir': working_dir}
        raise Exception('No site with a Globus endpoint and working_dir defined')

    def _get_working_dir(self, site_name):
        for s in self.config['sites']:
            if s['site'] == site_name:
                if 'data' not in s:
                    break
                data = s['data']
                if 'working_dir' not in data:
                    break
                return os.path.normpath(data['working_dir'])
        return None

    def stage_in(self, file, site_name):
        """Transport the file from the site of origin to the site.

        This function returns a DataFuture.

        Args:
            - self
            - file (File) : file to stage in
            - site_name (str|List) : a site the file is going to be staged in to.
                                If the site argument is not specified for a file
                                with 'globus' scheme, the file will be staged in to
                                the first site with the "globus" key in a config.
        """

        if file.scheme == 'file':
            stage_in_app = self._file_stage_in_app()
            app_fut = stage_in_app(outputs=[file])
            return app_fut._outputs[0]
        elif file.scheme == 'ftp':
            working_dir = self._get_working_dir(site_name)
            stage_in_app = self._ftp_stage_in_app(site_name=site_name)
            app_fut = stage_in_app(working_dir, outputs=[file])
            return app_fut._outputs[0]
        elif file.scheme == 'http' or file.scheme == 'https':
            working_dir = self._get_working_dir(site_name)
            stage_in_app = self._http_stage_in_app(site_name=site_name)<commit_msg>Fix documentation about DM.stagin_in() params<commit_after>import os
import logging
import requests
import ftplib
import concurrent.futures as cf
from parsl.executors.base import ParslExecutor
from parsl.data_provider.globus import get_globus
from parsl.app.app import App

logger = logging.getLogger(__name__)


class DataManager(ParslExecutor):
    """DataManager uses the familiar Executor interface, where staging tasks are submitted
    to it, and DataFutures are returned.
    """

    """
    In general a site where remote file is going to be stage in is unknown until
    Executer submits an app that depends on the file. However, in most practical
    cases, a site where an app is executed and a file needs to be staged in is
    known. Such cases should be detected by DataManager to optimize file
    transfers. Possible cases are:
    1. Config defines one site only.
    2. Config defines several sites but all of the sites share the filesystem /
       use the same Globus endpoint.
    3. Config defines several sites with different Globus endpoints but a user
       specified explicitely that apps must be executed on a particular site.
    """

    default_data_manager = None

    @classmethod
    def get_data_manager(cls, dfk=None, config=None):
        if not cls.default_data_manager or dfk:
            cls.default_data_manager = DataManager(dfk, config=config)
        return cls.default_data_manager

    def __init__(self, dfk, max_workers=10, config=None):
        """Initialize the DataManager.

        Kwargs:
           - max_workers (int) : Number of threads (Default=10)
           - config (dict): The config dict object for the site:

        """
        self._scaling_enabled = False
        self.executor = cf.ThreadPoolExecutor(max_workers=max_workers)

        self.dfk = dfk
        self.config = config
        if not self.config:
            self.config = {"sites": []}
        self.files = []
        self.globus = None

    def submit(self, *args, **kwargs):
        """Submit a staging app. All optimization should be here."""
        return self.executor.submit(*args, **kwargs)

    def scale_in(self, blocks, *args, **kwargs):
        pass

    def scale_out(self, *args, **kwargs):
        pass

    def shutdown(self, block=False):
        """Shutdown the ThreadPool.

        Kwargs:
            - block (Bool): To block for confirmations or not

        """
        x = self.executor.shutdown(wait=block)
        logger.debug("Done with executor shutdown")
        return x

    def scaling_enabled(self):
        return self._scaling_enabled

    def add_file(self, file):
        if file.scheme == 'globus':
            if not self.globus:
                self.globus = get_globus()
            # keep a list of all remote files for optimization purposes (TODO)
            self.files.append(file)
            self._set_local_path(file)

    def _set_local_path(self, file):
        globus_ep = self._get_globus_site()
        file.local_path = os.path.join(
                globus_ep['working_dir'],
                file.filename)

    def _get_globus_site(self, site_name='all'):
        if isinstance(site_name, list):
            site_name = site_name[0]
        for s in self.config['sites']:
            if site_name is 'all' or s['site'] == site_name:
                if 'data' not in s:
                    continue
                data = s['data']
                if 'globus' not in data or 'working_dir' not in data:
                    continue
                globus_ep = data['globus']
                if 'endpoint_name' not in globus_ep:
                    continue
                endpoint_name = data['globus']['endpoint_name']
                working_dir = os.path.normpath(data['working_dir'])
                if 'endpoint_path' in globus_ep and 'local_path' in globus_ep:
                    endpoint_path = os.path.normpath(globus_ep['endpoint_path'])
                    local_path = os.path.normpath(globus_ep['local_path'])
                    common_path = os.path.commonpath((local_path, working_dir))
                    if local_path != common_path:
                        raise Exception('"local_path" must be equal or an absolute subpath of "working_dir"')
                    relative_path = os.path.relpath(working_dir, common_path)
                    endpoint_path = os.path.join(endpoint_path, relative_path)
                else:
                    endpoint_path = working_dir
                return {'site_name': s['site'],
                        'endpoint_name': endpoint_name,
                        'endpoint_path': endpoint_path,
                        'working_dir': working_dir}
        raise Exception('No site with a Globus endpoint and working_dir defined')

    def _get_working_dir(self, site_name):
        for s in self.config['sites']:
            if s['site'] == site_name:
                if 'data' not in s:
                    break
                data = s['data']
                if 'working_dir' not in data:
                    break
                return os.path.normpath(data['working_dir'])
        return None

    def stage_in(self, file, site_name):
        """Transport the file from the site of origin to the site.

        This function returns a DataFuture.

        Args:
            - self
            - file (File) : file to stage in
            - site_name (str) : a site the file is going to be staged in to.
                                If the site argument is not specified for a file
                                with 'globus' scheme, the file will be staged in to
                                the first site with the "globus" key in a config.
        """

        if file.scheme == 'file':
            stage_in_app = self._file_stage_in_app()
            app_fut = stage_in_app(outputs=[file])
            return app_fut._outputs[0]
        elif file.scheme == 'ftp':
            working_dir = self._get_working_dir(site_name)
            stage_in_app = self._ftp_stage_in_app(site_name=site_name)
            app_fut = stage_in_app(working_dir, outputs=[file])
            return app_fut._outputs[0]
        elif file.scheme == 'http' or file.scheme == 'https':
            working_dir = self._get_working_dir(site_name)
            stage_in_app = self._http_stage_in_app(site_name=site_name)
{'fraction_non_alphanumeric': 0.0821549885456987, 'fraction_numerical': 0.0015799036258788213, 'mean_word_length': 1.9774223894637817, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, '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': '3934580', 'n_tokens_mistral': 3540, 'n_tokens_neox': 3383, 'n_words': 1071}