KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
