STUDIOML_EXPERIMENT = 'STUDIOML_EXPERIMENT'
STUDIOML_HOME = 'STUDIOML_HOME'
STUDIOML_ARTIFACT_MAPPING = 'STUDIOML_ARTIFACT_MAPPING'
def get_experiment_key():
def get_studio_home():
def _setup_model_directory(experiment_name, clean=False):
def setup_experiment(env, experiment, clean=True):
def get_artifact(tag):
def get_artifacts():
def get_artifact_cache(tag, experiment_name=None):
def get_blob_cache(blobkey):
def _get_artifact_mapping_path(experiment_name=None):
def get_tensorboard_dir(experiment_name=None): DAY = 86400
HOUR = 3600
MINUTE = 60
def remove_backspaces(line):
def report_fatal(msg: str, logger):
def sha256_checksum(filename, block_size=65536):
def filehash(filename, block_size=65536, hashobj=None):
def rand_string(length):
def _looks_like_url(name):
def parse_s3_path(qualified: str):
def retry(f,
          no_retries=5, sleep_time=1,
          exception_class=BaseException, logger=None):
def compression_to_extension(compression):
def compression_to_taropt(compression):
def _compression_to_extension_taropt(compression):
def timeit(method):
    def timed(*args, **kw):
def sixdecode(sval):
def shquote(sval):
def parse_duration(duration_str):
def parse_verbosity(verbosity=None):
def str2duration(sval):
def get_temp_filename() -> str:
def delete_local_path(local_path: str, root: str, shallow: bool):
def delete_local_folders(local_folder_path: str, root: str):
def rm_rf(path):
def check_for_kb_interrupt():
def add_packages(list1, list2): DEBUG = logging.DEBUG
INFO = logging.INFO
ERROR = logging.ERROR
def get_logger(name):
def debug(line):
def error(line):
def info(line): AWS_KEY = 'access_key'
AWS_SECRET_KEY = 'secret_access_key'
AWS_SESSION_TOKEN = 'session_token'
AWS_REGION = 'region'
AWS_PROFILE = 'profile'
KEY_CREDENTIALS = 'credentials'
AWS_TYPE = 'aws'
class Credentials:
    def __init__(self, cred_dict):
    def get_type(self):
    def get_key(self):
    def get_secret_key(self):
    def get_session_token(self):
    def get_region(self):
    def get_profile(self):
    def _get_named(self, key: str, config: Dict):
    def to_dict(self):
    def get_fingerprint(self) -> str:
    def get_credentials(cls, config):
    def to_string(self): DB_KEY = "database"
STORE_KEY = "store"
def setup_storage(db_provider, artifact_store):
def get_storage_db_provider():
def get_storage_artifact_store():
def reset_storage():
def get_storage_verbose_level():
def set_storage_verbose_level(level: int): class StorageType(Enum):
    storageHTTP = 1
    storageS3 = 2
    storageLocal = 3
    storageFirebase = 4
    storageDockerHub = 5
    storageSHub = 6

    storageInvalid = 99 class StorageHandler:
    def __init__(self, storage_type: StorageType,
                 logger,
                 measure_timestamp_diff=False,
                 compression=None):
        self.type = storage_type
        self.logger = logger
        self.compression = compression
        self._timestamp_shift = 0
        if measure_timestamp_diff:
            try:
                self._timestamp_shift = self._measure_timestamp_diff()
            except BaseException:
                check_for_kb_interrupt()
                self._timestamp_shift = 0

    @classmethod
    def get_id(cls, config: Dict) -> str:
        raise NotImplementedError("Not implemented: upload_file")

    def upload_file(self, key, local_path):
        raise NotImplementedError("Not implemented: upload_file")

    def download_file(self, key, local_path):
        raise NotImplementedError("Not implemented: download_file")

    def download_remote_path(self, remote_path, local_path):
        raise NotImplementedError("Not implemented: download_remote_path")

    def delete_file(self, key, shallow=True):
        raise NotImplementedError("Not implemented: delete_file")

    def get_file_url(self, key, method='GET'):
        raise NotImplementedError("Not implemented: get_file_url")

    def get_file_timestamp(self, key):
        raise NotImplementedError("Not implemented: get_file_timestamp")

    def get_qualified_location(self, key):
        raise NotImplementedError("Not implemented: get_qualified_location")

    def get_local_destination(self, remote_path: str):
        raise NotImplementedError("Not implemented: get_local_destination")

    def get_timestamp_shift(self):
        return self._timestamp_shift

    def get_compression(self):
        return self.compression

    def cleanup(self):
        pass

    def _measure_timestamp_diff(self):
        max_diff = 60
        tmpfile = get_temp_filename() + '.txt'
        with open(tmpfile, 'w') as f:
            f.write('timestamp_diff_test')
        key = 'tests/' + str(uuid.uuid4())
        self.upload_file(key, tmpfile)
        remote_timestamp = self.get_file_timestamp(key)

        if remote_timestamp is not None:
            now_remote_diff = time.time() - remote_timestamp
            self.storage_handler.delete_file(key)
            os.remove(tmpfile)

            assert -max_diff < now_remote_diff and \
                now_remote_diff < max_diff, \
                "Timestamp difference is more than 60 seconds. " + \
                "You'll need to adjust local clock for caching " + \
                "to work correctly"

            return -now_remote_diff if now_remote_diff < 0 else 0 class StorageHandlerFactory:
    def __init__(self):
        self.logger = logs.get_logger(self.__class__.__name__)
        self.logger.setLevel(get_storage_verbose_level())
        self.handlers_cache = dict()
        self.cleanup_at_exit: bool = True

    @classmethod
    def get_factory(cls):
        global _storage_factory
        if _storage_factory is None:
            _storage_factory = StorageHandlerFactory()
        return _storage_factory

    def set_cleanup_at_exit(self, value: bool):
        self.cleanup_at_exit = value

    def cleanup(self):
        if not self.cleanup_at_exit:
            return

        for _, handler in self.handlers_cache.items():
            if handler is not None:
                handler.cleanup()

    def get_handler(self, handler_type: StorageType,
                    config: Dict) -> StorageHandler:
        if handler_type == StorageType.storageS3:
            handler_id: str = S3StorageHandler.get_id(config)
            handler = self.handlers_cache.get(handler_id, None)
            if handler is None:
                handler = S3StorageHandler(config)
                self.handlers_cache[handler_id] = handler
            return handler
        if handler_type == StorageType.storageHTTP:
            handler_id: str = HTTPStorageHandler.get_id(config)
            handler = self.handlers_cache.get(handler_id, None)
            if handler is None:
                handler = HTTPStorageHandler(
                    config.get('endpoint', None),
                    config.get('credentials', None))
                self.handlers_cache[handler_id] = handler
            return handler
        if handler_type == StorageType.storageLocal:
            handler_id: str = LocalStorageHandler.get_id(config)
            handler = self.handlers_cache.get(handler_id, None)
            if handler is None:
                handler = LocalStorageHandler(config)
                self.handlers_cache[handler_id] = handler
            return handler
        self.logger("FAILED to get storage handler: unsupported type %s",
                    repr(handler_type))
        return None def tar_artifact(local_path: str, key: str,
                 compression: str, logger, cache: bool = True):
    tar_filename: str = util.get_temp_filename() + ".tar"

    if local_path != '/' and local_path.endswith('/'):
        local_path = local_path[0..len(local_path)-1]

    #TODO: ASD process ignore files list if present:
    ignore_filepath = _find_ignore_list(local_path)
    if ignore_filepath is not None:
        logger.info("NOT PROCESSED: ignore files list %s for artifact %s.",
                    ignore_filepath, local_path)

    if cache and key:
        cache_path = artifacts_tracker.get_artifact_cache(key)
        if cache_path != local_path:
            msg: str =\
                "NOT IMPLEMENTED: artifact local path and cache location differ: {0} vs {1}."\
                    .format(local_path, cache_path)
            logger.error(msg)
            raise NotImplementedError(msg)

    tic = time.time()
    if os.path.isdir(local_path):
        _tar_artifact_directory(local_path, tar_filename,
                                key, ignore_filepath, logger)
    else:
        _tar_artifact_single_file(local_path, tar_filename,
                                  key, logger)
    toc = time.time()

    logger.debug('tar finished in %f s', (toc - tic))
    return tar_filename def untar_artifact(local_path: str, tar_filename: str, logger):

    if local_path != '/' and local_path.endswith('/'):
        local_path = local_path[0..len(local_path)-1]

    logger.debug("Untarring %s", tar_filename)

    tf = None
    try:
        tf = tarfile.open(tar_filename, 'r')
        tar_items = tf.getnames()
        logger.debug('List of files in the tar: ' + str(tar_items))

        single_file_name = _get_single_file_name(tar_items)
        if single_file_name is None:
            util.rm_rf(local_path)
            # Extract tar file into directory "local_path"
            os.makedirs(local_path, exist_ok=True)
            logger.debug("Untarring %s to dir. %s", tar_filename, local_path)
            tf.extractall(local_path)
        else:
            with tempfile.TemporaryDirectory() as temp_dir:
                logger.debug("Untarring %s to temp. dir. %s", tar_filename, temp_dir)
                tf.extractall(temp_dir)
                util.rm_rf(local_path)
                temp_path: str = os.path.join(temp_dir, single_file_name)
                logger.debug("Copying single file %s to %s", temp_path, local_path)
                shutil.copy2(temp_path, local_path)

    except Exception as exc:
        msg: str = \
            "FAILED to extract tarfile: {0} for artifact {1} reason: {2}" \
                .format(tar_filename, local_path, exc)
        logger.error(msg)
    finally:
        if tf is not None:
            tf.close()
import hashlib
import os
import re
import tarfile
    from urllib.request import urlopen
    from urllib import urlopen
from typing import Dict
from studio.artifacts import artifacts_tracker
from studio.util import util, logs
from studio.credentials import credentials
from studio.storage import storage_setup
from studio.storage.storage_type import StorageType
from studio.storage.storage_handler import StorageHandler
from studio.storage.storage_handler_factory import StorageHandlerFactory
from studio.storage.storage_util import tar_artifact, untar_artifact

try:
except ImportError:


# The purpose of this class is to encapsulate the logic
# of handling artifact's state and it's transition between
# being local on client's side, cached in shared storage location,
# downloaded into payload execution environment etc.
# Part of artifact is a reference to StorageHandler instance,
# potentially unique for each Artifact instance.
# This StorageHandler defines where artifact is currently stored
# and how it is accessed.
class Artifact:

    def __init__(self, art_name, art_dict, logger=None):
        self.name = art_name
        self.key: str = None
        self.local_path: str = None
        self.remote_path: str = None
        self.credentials = None
        self.hash = None

        self.logger = logger
        if self.logger is None:
            self.logger = logs.get_logger(self.__class__.__name__)
            self.logger.setLevel(storage_setup.get_storage_verbose_level())

        self.storage_handler: StorageHandler = None

        self.unpack: bool = art_dict.get('unpack')
        self.is_mutable: bool = art_dict.get('mutable')
        if 'key' in art_dict.keys():
            self.key = art_dict['key']
        if 'local' in art_dict.keys():
            self.local_path = art_dict['local']
        if 'qualified' in art_dict.keys():
            self.remote_path = art_dict['qualified']
        if 'url' in art_dict.keys():
            self.remote_path = art_dict['url']
        if 'hash' in art_dict.keys():
            self.hash = art_dict['hash']
        self.credentials = credentials.Credentials.get_credentials(art_dict)

        self._setup_storage_handler(art_dict)

    def upload(self, local_path=None):
        if self.storage_handler is None:
            msg: str = "No storage handler is set for artifact {0}"\
                .format(self.key)

            util.report_fatal(msg, self.logger)