KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use cache_dirs method in avocado

Best Python code snippet using avocado_python

titanic-starter.py

Source:titanic-starter.py

...430    tmp.columns = ['model_name', 'train', 'test']431    tmp_sorted = tmp.sort_values(['test'], ascending=[False])432    ICD.display(tmp_sorted.head(limit))433# In[ ]:434def clean_cache_dirs(cache_dirs):435    for cache_dir in cache_dirs:436        if os.path.exists(cache_dir):437            rmtree(cache_dir)438    cache_dirs = []439# In[ ]:440def build_pipelline(base_model, pipeline, param_grid, reduce_dim_param_grid, grid_search_parameters, cache_dirs):441    param_grid = [{'classify__'+key: value for (key, value) in inner_param_grid.items()} for inner_param_grid in param_grid]442    if type(param_grid) == list:443        for i in range(len(param_grid)):444            for key, value in reduce_dim_param_grid.items():445                param_grid[i][key] = value446    if type(param_grid) == dict:447        for key, value in reduce_dim_param_grid.items():448            param_grid[key] = value449    cachedir = mkdtemp()450    cache_dirs.append(cachedir)451    memory = Memory(cachedir=cachedir, verbose=0)452    pipe = Pipeline(pipeline, memory=memory)453    gridsearch = GridSearchCV(454        pipe, 455        param_grid = param_grid, 456        **grid_search_parameters457    )458    return gridsearch459def build_experiments(model_name, base_model, param_grid, grid_search_parameters, cache_dirs):460    461    reduce_dims = [462        None,463        SelectKBest(chi2),464        SelectPercentile(chi2),465        SelectFromModel(LinearSVC(penalty="l1", dual=False))466    ]467    468    scalers = [469        None,470        MinMaxScaler(),471        RobustScaler(quantile_range=(25, 75)),472        Normalizer()473    ]474    475    experiments = []476    477    for reduce_dim in reduce_dims:478        for scaler in scalers:479            experiment_name = model_name480            pipeline = []            481            if reduce_dim != None:482                pipeline.append(('reduce_dim', reduce_dim))483                experiment_name += type(reduce_dim).__name__484            if scaler != None:485                pipeline.append(('scaler', scaler))486                experiment_name += type(scaler).__name__487            pipeline.append(('classify', clone(base_model)))488            experiment = build_pipelline(489                base_model, 490                pipeline , 491                param_grid, 492                {}, 493                grid_search_parameters, 494                cache_dirs495            )496            experiments.append((experiment_name, experiment))497    498    #if hasattr(base_model, 'coef_') or hasattr(base_model, 'feature_importances_'):499    #    rfecv = RFECV(estimator=clone(base_model), step=1, cv=StratifiedKFold(2), scoring='accuracy')500    #    experiments.append((model_name + '-rfecv', rfecv))501    502    return experiments503# In[ ]:504def run_experiments(train_X , train_y, evaluate_X, evaluate_y, experiments, models, evaluation, feature_selections, scores, cache_dirs):505    for (model_name, model) in experiments:506        print('Fit', model_name)507        model.fit(train_X , train_y)508        print('model', type(model).__name__)509        if type(model).__name__ == 'GridSearchCV':510            best_model = model.best_estimator_511            print(type(best_model))512            show_search(model)513            if type(best_model).__name__ == 'Pipeline':514                if 'reduce_dim' in best_model.named_steps:515                    reduce_dim = best_model.named_steps['reduce_dim']516                    selected_features = print_features(reduce_dim)517                    feature_selections[model_name] = [518                        selected_features,519                        model.best_score_520                    ]521        elif type(model).__name__ == 'RFECV':522            plot_rfecv(model)523            best_model = model524            selected_features = print_rfecv_features(best_model)525            feature_selections[model_name] = [526                selected_features,527                model.best_score_528            ]529        else:530            best_model = model531        532        print('best_model', type(best_model).__name__)533        models[model_name] = {}534        models[model_name][scoring_refit] = best_model535        evaluation[model_name] = evaluate_model(best_model, evaluate_X, evaluate_y)536        scores[model_name] = (best_model.score(train_X, train_y), best_model.score(evaluate_X, evaluate_y))537    clean_cache_dirs(cache_dirs)538# ## Model Parameter Search539# ### RandomForestClassifier540# In[ ]:541#reduced_X = 542#reduce_dim = models['RandomForestClassifier'][scoring_refit].named_steps['reduce_dim']543#reduced_X = pd.DataFrame(reduce_dim.transform(train_X), columns=train_X.columns[reduce_dim.get_support()])544#best_model = models['RandomForestClassifier'][scoring_refit].named_steps['classify']545model_name = 'RandomForestClassifier'546base_model = RandomForestClassifier(random_state=0)547param_grid = [548    {549        "max_features" : ['sqrt', 'log2'],550        "max_depth": [None, 1, 5, 10, 100],551        "n_estimators" :[10, 100, 1000],...

clone.py

Source:clone.py

1import os2import sys3import git4import shutil5from threading import Thread6class DownloadGit(Thread):7    """8    """9    def __init__(self, url, repo_name, destination):10        Thread.__init__(self)11        self.url = url12        self.repo_name = repo_name13        self.destination = destination14    def run(self):15        """16        :return:17        """18        cache_dir = os.getcwd() + '/cache/'19        current_full_paths = map(lambda name: os.path.join(self.destination, name), os.listdir(self.destination))20        cache_full_paths = map(lambda name: os.path.join(cache_dir, name), os.listdir(cache_dir))21        current_dirs = []22        cache_dirs = []23        for path in current_full_paths:24            if os.path.isdir(path):25                current_dirs.append(os.path.basename(path))26        for path in cache_full_paths:27            if os.path.isdir(path):28                cache_dirs.append(os.path.basename(path))29        if self.repo_name in current_dirs and self.repo_name in cache_dirs:30            pass31        elif self.repo_name not in current_dirs and self.repo_name in cache_dirs:32            cache_repo = f'{self.destination}/cache/{self.repo_name}'33            repo = git.Repo(cache_repo)34            origin = repo.remotes.origin35            origin.pull()36            shutil.copytree(37                cache_repo,38                f'{self.destination}/{self.repo_name}',39                dirs_exist_ok=True40            )41        else:42            cache = f'{self.destination}/cache/{self.repo_name}'43            git.Repo.clone_from(url=self.url, to_path=cache)44            destination = f'{self.destination}/{self.repo_name}'45            shutil.copytree(cache, destination, dirs_exist_ok=True)46def main(url, destination):47    """48    """49    repo_name = os.path.basename(url)50    thread = DownloadGit(url, repo_name, destination)51    thread.start()52if __name__ == "__main__":53    url = sys.argv[1]54    destination = sys.argv[2]55    # url = 'https://github.com/DennTerentyev/Dockerfiles'56    # destination = os.getcwd()...

cache.py

Source:cache.py

1import glob2import hashlib3import json4from pathlib import Path5from pyknife.aws import S3File6def find_cache(hash_val,cache_path):7	if cache_path.startswith('s3://'):8		s3_obj = S3File(cache_path,hash_val)9		#Maybe cache path was already downloaded from S3, in that case first look in locals10		cache_dir = Path(cache_path.split('s3://')[-1],hash_val)11		cache_dirs = glob.glob(str(cache_dir.absolute())+'_*/*')12		if len(cache_dirs)>0:13			return cache_dirs14		else:15			#If no cache found local, look in S316			cache_dirs = s3_obj.glob(s3_obj.path+'_*/*')17			if len(cache_dirs)>0:18				return cache_dirs19			else:20				return None21	else:22		cache_dir = Path(cache_path,hash_val)23		cache_dirs = glob.glob(str(cache_dir.absolute())+'_*/*')24	if len(cache_dirs) > 0:25		return cache_dirs26	else:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.