Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use prepare_data_args method in autotest

Best Python code snippet using autotest_python

app.py

Source:app.py

1import dubhe_sdk2from geek_vision_algo.train import main as model_train3from geek_vision_algo.predict import main as model_predict4from geek_vision_algo.inference import main as model_inference5import torchvision.transforms as transforms6import torchvision.datasets as datasets7from torch.utils.data import random_split8import glob9import re10import shutil11import traceback12import json13import os14import pandas as pd15import numpy as np16from dataset.pretrain_prepare import main as prepare_data17from dataset import MyDataset18import torch19import geek_vision_algo.model as mm20def load_config(config_file):21    try:22        with open(config_file) as file:23            config =json.load(file)24            return config25    except OSError as e:26        print(e)27        return None28@dubhe_sdk.pipeline()29class Train_Model():30    def __init__(self, ctx:dubhe_sdk.Context):31        config_params = load_config('config/model_train_config.json')32        self.model_name = config_params['model']33        self.evaluation_interval = config_params['evaluation_interval']34        self.epochs = config_params['epochs']35        self.batch_size = config_params['batch_size']36        self.lr = config_params['lr']37    @dubhe_sdk.train()38    def train(self, ctx: dubhe_sdk.Context):39        # get input parameters40        self.data_paths = ctx.get(dubhe_sdk.ContextKey.KEY_DATA_PATHS)41        self.code_list = ctx.get(dubhe_sdk.ContextKey.KEY_CODE_LIST)42        self.pretrained_model_path = ctx.get(dubhe_sdk.ContextKey.KEY_PRETRAINED_MODEL_PATH)43        self.pretrained_epoch = ctx.get(dubhe_sdk.ContextKey.KEY_PRETRAINED_EPOCH)44        self.output_dir = ctx.get(dubhe_sdk.ContextKey.KEY_OUTPUT_DIR)45        self.debug_dir = ctx.get(dubhe_sdk.ContextKey.KEY_TMP_DIR)46        self.split_rate = ctx.get(dubhe_sdk.ContextKey.KEY_SPLIT_RATE)47        self.platform_type = ctx.get(dubhe_sdk.ContextKey.ENV_PLATFORM_TYPE)48        # choose gpu to train49        if (self.platform_type != 1):50            os.environ["CUDA_VISIBLE_DEVICES"] = "0"51        # prepeare dir52        if not os.path.exists(self.debug_dir):53            os.makedirs(self.debug_dir, exist_ok=True)54        if not os.path.exists(self.output_dir):55            os.makedirs(self.output_dir, exist_ok=True)56        # split train and test57        prepare_data_args = [58            "--data_paths",59            *self.data_paths,60            "--code_list",61            *self.code_list,62            "--val_ratio=%s"%self.split_rate,63            "--debug_dir=%s"%self.debug_dir,64            '--predict_method=train',65        ]66        prepare_data(prepare_data_args)67        df_train = pd.read_csv(os.path.join(self.debug_dir, "train.csv"))68        df_valid = pd.read_csv(os.path.join(self.debug_dir, "val.csv"))69        code_count_dict = dict(zip(*np.unique(df_train["code"].values.tolist(), return_counts=True)))70        code_count_list = sorted(code_count_dict, key=lambda i:i[1], reverse=True)71        code2label = {}72        for idx, code in enumerate(code_count_list):73            code2label[code] = idx74        ctx.log(f'Code list used for training is  {code_count_list}.')75        normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],76                                         std=[0.229, 0.224, 0.225])77        train_transforms = transforms.Compose([78            transforms.Resize((224, 224)),79            transforms.RandomHorizontalFlip(p=0),80            transforms.ColorJitter(brightness=0, contrast=0, saturation=0,81                                   hue=0),82            transforms.ToTensor(),83            normalize,84        ])85        valid_transforms = transforms.Compose([86            transforms.Resize((224, 224)),87            transforms.ToTensor(),88            normalize,89        ])90        train_dataset = MyDataset(df_train,91                                  code2label,92                                  train_transforms)93        valid_dataset = MyDataset(df_valid,94                                  code2label,95                                  valid_transforms)96        self.train_dataset = train_dataset97        self.valid_dataset = valid_dataset98        # train model99        ctx.log('model train starting...')100        train_args = [101            '--model_name=%s' % self.model_name,102            '--init_epoch=%d' % self.pretrained_epoch,103            '--epochs=%d' % self.epochs,104            '--batch_size=%d' % self.batch_size,105            '--lr=%f' % self.lr,106            '--evaluation_interval=%d' % self.evaluation_interval,107            '--save_path=%s' % self.output_dir,108            '--progress_init=0.0',109            '--progress_ratio=0.9',110            '--model_optimizer=SGD',111            '--code_list',112            *code_count_list,113            '--model_num=0'114        ]115        ctx.log('model train args \n %s ' % train_args)116        model_train(train_args, self.train_dataset, self.valid_dataset, ctx)  # please design your model train logic in this function.117        ctx.log('model train end.')118        # 2. find best model file119        model_files = glob.glob(os.path.join(self.output_dir, '*'))120        model_prefix = r'model_(\d)+_{}'.format(self.epochs - 1)121        for file in model_files:122            model_name = os.path.basename(file)123            if re.match(model_prefix, model_name, re.M | re.I):124                self.model_file = model_name125                break126        # 3. predict model127        ctx.log('model predict starting...')128        self.output_txt = os.path.join(self.debug_dir, 'pred_result.txt')129        self.output_img = os.path.join(self.debug_dir, 'vis')130        model_path = os.path.join(self.output_dir, self.model_file)131        if model_path:132            model_dict = torch.load(model_path)133            self.state_dict = model_dict['state_dict']134        predict_args = [135            '--model_name=%s' % self.model_name,136            '--output_txt=%s' % self.output_txt,137            '--output_img=%s' % self.output_img,138            '--batch_size=%d' % self.batch_size,139            '--progress_init=0.9',140            '--progress_ratio=0.1',141            '--code_list',142            *code_count_list,143            '--model_num=0'144        ]145        ctx.log('model predict args \n %s' % predict_args)146        model_predict(predict_args, self.valid_dataset, self.state_dict, ctx)147        ctx.log('model predict end.')148@dubhe_sdk.pipeline()149class Predict_Model():150    def __init__(self, ctx:dubhe_sdk.Context):151        config_params = load_config('config/model_predict_config.json')152        self.model_name = config_params['model']153        self.batch_size = config_params['batch_size']154        # ç¦»çº¿è¯ä¼°å¯å¨å è½½æ¨¡å155        model_path = ctx.get(dubhe_sdk.ContextKey.ENV_MODEL_PATH)156        model_dict = torch.load(model_path)157        self.code_list = model_dict['code_list']158        self.state_dict = model_dict['state_dict']159    @dubhe_sdk.predict()160    def predict(self, ctx: dubhe_sdk.ContextKey):161        self.data_paths = ctx.get(dubhe_sdk.ContextKey.KEY_DATA_PATHS)162        self.debug_dir = ctx.get(dubhe_sdk.ContextKey.KEY_TMP_DIR)163        ctx.log('model predict starting...')164        # prepeare dir165        if not os.path.exists(self.debug_dir):166            os.makedirs(self.debug_dir, exist_ok=True)167        normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],168                                         std=[0.229, 0.224, 0.225])169        test_transforms = transforms.Compose([170            transforms.Resize((224, 224)),171            transforms.ToTensor(),172            normalize,173        ])174        prepare_data_args = [175            "--data_paths",176            *self.data_paths,177            "--debug_dir=%s" % self.debug_dir,178            '--predict_method=predict',179        ]180        prepare_data(prepare_data_args)181        df_test = pd.read_csv(os.path.join(self.debug_dir, "test.csv"))182        code2label = {}183        for idx, code in enumerate(self.code_list):184            code2label[code] = idx185        test_dataset = MyDataset(df_test,186                                  code2label,187                                  test_transforms)188        self.output_txt = os.path.join(self.debug_dir, 'pred_result.txt')189        self.output_img = os.path.join(self.debug_dir, 'vis')190        predict_args = [191            '--model_name=%s' % self.model_name,192            '--output_txt=%s' % self.output_txt,193            '--output_img=%s' % self.output_img,194            '--batch_size=%d' % self.batch_size,195            '--progress_init=0.9',196            '--progress_ratio=0.1',197            '--code_list',198            *self.code_list,199            '--model_num=0'200        ]201        ctx.log('model predict args \n %s' % predict_args)202        model_predict(predict_args, test_dataset, self.state_dict, ctx)203        ctx.log('model predict end.')204@dubhe_sdk.pipeline()205class Infer_Model():206    def __init__(self, ctx: dubhe_sdk.Context):207        config_params = load_config('config/model_inference_config.json')208        self.batch_size = config_params['batch_size']209        self.model_name = config_params["Gechuang_aimidware_model_base_or_full_name"]210        # å¨çº¿æ¨çå¯å¨å è½½æ¨¡å211        rely_model_data = ctx.get(dubhe_sdk.ContextKey.ENV_RELY_MODEL_DATA)212        model_info = rely_model_data[self.model_name]213        ckpt_dir = model_info['model_path']214        ctx.log('Model params has been loaded from %s.' % ckpt_dir)215        model_dict = torch.load(ckpt_dir)216        self.code_list = model_dict['code_list']217        self.state_dict = model_dict['state_dict']218        resnet = getattr(mm, config_params["model"])()219        self.model = mm.Net(resnet, config_params["model"],len(self.code_list))220        self.model.load_state_dict(self.state_dict)221        self.model.eval()222        if torch.cuda.is_available():223            self.model.cuda()224    @dubhe_sdk.inference()225    def inference(self, ctx:dubhe_sdk.ContextKey):226        image_base_dir = ctx.get(dubhe_sdk.ContextKey.KEY_IMAGE_BASE_DIR)227        image_names = ctx.get(dubhe_sdk.ContextKey.KEY_IMAGE_NAMES)228        self.debug_dir = ctx.get(dubhe_sdk.ContextKey.KEY_TMP_DIR)229        ctx.log('model inference starting...')230        # prepeare dir231        if not os.path.exists(self.debug_dir):232            os.makedirs(self.debug_dir, exist_ok=True)233        full_image_names = [os.path.join(image_base_dir, image_name) for image_name in image_names]234        df_total = pd.DataFrame()235        for img_cache in full_image_names:236            df_total_cache = pd.DataFrame()237            df_total_cache['image'] = [img_cache]238            df_total_cache['code'] = image_base_dir.split('/')[-1]239            df_total = pd.concat([df_total, df_total_cache])240        ctx.log(f'The data set shape is {df_total.shape}.')241        normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],242                                         std=[0.229, 0.224, 0.225])243        test_transforms = transforms.Compose([244            transforms.Resize((224, 224)),245            transforms.ToTensor(),246            normalize,247        ])248        code2label = {}249        for idx, code in enumerate(self.code_list):250            code2label[code] = idx251        test_dataset = MyDataset(df_total,252                                 code2label,253                                 test_transforms)254        self.output_txt = os.path.join(self.debug_dir, 'pred_result.txt')255        self.output_img = os.path.join(self.debug_dir, 'vis')256        predict_args = [257            '--output_txt=%s' % self.output_txt,258            '--output_img=%s' % self.output_img,259            '--batch_size=%d' % self.batch_size,260            '--progress_init=0.9',261            '--progress_ratio=0.1',262            '--code_list',263            *self.code_list,264            '--model_num=0'265        ]266        ctx.log('model inference args \n %s' % predict_args)267        data = model_inference(predict_args, test_dataset, self.model, ctx)268        ctx.log('model inference end.')...

rdb_model_extensions.py

Source:rdb_model_extensions.py

...86            if field_name not in field_dict:87                errors[field_name] = 'No field of this name'88        return errors89    @classmethod90    def prepare_data_args(cls, data):91        'Common preparation for add_object and update_object'92        # must check for extraneous field names here, while we have the93        # data in a dict94        errors = cls.validate_field_names(data)95        if errors:96            raise django_exceptions.ValidationError(errors)97        return data98    @classmethod99    def _get_required_field_names(cls):100        """Get the fields without which we cannot create a host.101        @return: A list of field names that cannot be blank on host creation.102        """103        return [field.name for field in cls._meta.fields if not field.blank]104    @classmethod105    def get_basic_field_names(cls):106        """Get all basic fields of the Model.107        This method returns the names of all fields that the client can provide108        a value for during host creation. The fields not included in this list109        are those that we can leave blank. Specifying non-null values for such110        fields only makes sense as an update to the host.111        @return A list of basic fields.112            Eg: set([hostname, locked, leased, status, invalid,113                     protection, lock_time, dirty])114        """115        return [field.name for field in cls._meta.fields116                if field.has_default()] + cls._get_required_field_names()117    @classmethod118    def validate_model_fields(cls, data):119        """Validate parameters needed to create a host.120        Check that all required fields are specified, that specified fields121        are actual model values, and provide defaults for the unspecified122        but unrequired fields.123        @param dict: A dictionary with the args to create the model.124        @raises dajngo_exceptions.ValidationError: If either an invalid field125            is specified or a required field is missing.126        """127        missing_fields = set(cls._get_required_field_names()) - set(data.keys())128        if missing_fields:129            raise django_exceptions.ValidationError('%s required to create %s, '130                    'supplied %s ' % (missing_fields, cls.__name__, data))131        data = cls.prepare_data_args(data)132        data = cls.provide_default_values(data)133        return data134class AbstractHostModel(dbmodels.Model, ModelValidators):135    """Abstract model specifying all fields one can use to create a host.136    This model enforces consistency between the host models of the rdb and137    their representation on the client side.138    Internal fields:139        status: string describing status of host140        invalid: true if the host has been deleted141        protection: indicates what can be done to this host during repair142        lock_time: DateTime at which the host was locked143        dirty: true if the host has been used without being rebooted144        lock_reason: The reason for locking the host.145    """...

test_qserv_cli.py

Source:test_qserv_cli.py

...109        remove=ANY,110    )111    args.update(kwargs)112    return args113def prepare_data_args(**kwargs):114    """Get a safely-mutable dict of the launch.launch kwargs, with all the values set to ANY,115    for use with mock.assert_called_with().116    Parameters117    ----------118    **kwargs : dict [`str`: `Any`]119        keyword argumets to override existing argments, or add new arguments if120        applicable.121    Returns122    -------123    arguments : `dict` [`str` : Union[`unittest.mock.ANY`, `Any`]]124        A dict of arguments that can be changed with expected values.125    """126    args = dict(127        qserv_root=ANY,128        itest_container=ANY,129        qserv_image=ANY,130        itest_file=ANY,131        outdir=ANY,132        dry=ANY,133        project=ANY,134    )135    args.update(kwargs)136    return args137class QservCliTestCase(unittest.TestCase):138    """Tests features of the qserv command line interface."""139    def setUp(self):140        self.runner = CliRunner()141    @patch.object(launch, "build")142    def test_EnvVal_var(self, build_mock):143        """Verify that an `opt.FlagEnvVal` option can be set using the environment144        variable.145        """146        fake_root = "/foo/bar/qserv"147        res = self.runner.invoke(qserv, ["build"], env=env_without_qserv(QSERV_ROOT=fake_root))148        self.assertEqual(res.exit_code, 0)149        build_mock.assert_called_once()150        build_mock.assert_called_with(**build_args(qserv_root=fake_root))151    @patch.object(launch, "build")152    def test_EnvVal_flag(self, build_mock):153        """Verify that an `opt.FlagEnvVal` option can be set using the flag and it154        overrides the environment variable.155        """156        fake_root = "/foo/bar/qserv"157        flag_root = "/path/to/qserv"158        res = self.runner.invoke(159            qserv,160            ["build", "--qserv-root", flag_root],161            env=env_without_qserv(QSERV_ROOT=fake_root),162        )163        self.assertEqual(res.exit_code, 0)164        build_mock.assert_called_once()165        build_mock.assert_called_with(**build_args(qserv_root=flag_root))166    @patch.object(launch, "itest", return_value=0)167    def test_itest_OptDefault_default(self, itest_mock):168        """Verify that an `opt.OptDefault` option value can be inferred from its169        associated environment variable.170        """171        res = self.runner.invoke(qserv, ["itest"], env=env_without_qserv())172        self.assertEqual(res.exit_code, 0)173        itest_mock.assert_called_once()174        expected = os.path.abspath(175            # This is the location of the path *inside* the build container,176            # because that's where the unit tests run. This is from the install177            # location of the unit tests back up the tree to the root qserv178            # folder.179            os.path.join(__file__, "../../../../../../../..")180        )181        itest_mock.assert_called_with(**itest_args(qserv_root=expected))182    @patch.object(launch, "prepare_data", return_value=0)183    def test_prepare_data_OptDefault_default(self, prepare_data_mock):184        """Verify that an `opt.OptDefault` option value can be inferred from its185        associated environment variable.186        """187        res = self.runner.invoke(qserv, ["prepare-data"], env=env_without_qserv())188        self.assertEqual(res.exit_code, 0)189        prepare_data_mock.assert_called_once()190        expected = os.path.abspath(191            # This is the location of the path *inside* the build container,192            # because that's where the unit tests run. This is from the install193            # location of the unit tests back up the tree to the root qserv194            # folder.195            os.path.join(__file__, "../../../../../../../..")196        )197        prepare_data_mock.assert_called_with(**prepare_data_args(qserv_root=expected))198    def test_env(self):199        """Test that `qserv env` runs without throwing.200        Does not check for correctness of course, but at least checks that201        values can be gotten and reported.202        """203        res = self.runner.invoke(qserv, ["env"])204        self.assertEqual(res.exit_code, 0)205if __name__ == "__main__":...

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.