Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _init_vars method in ATX

Best Python code snippet using ATX

GA_win_model.py

Source:GA_win_model.py

1# This is the GA process model done with machine learning2import framework.data.formats as formats3from framework.models.ml_model_memfix import MLModelMemfix4from copy import deepcopy5class GAWinModel:6    _META_DATA = [formats.UID, formats.PSEUDO_REP, formats.DATE]7    _INIT_DATA = [formats.ROW, formats.COL, formats.GENOTYPE,8                  formats.DD, formats.DOY, formats.PAR, formats.RAINFALL]9    _STAGE_ONE = [formats.TRANSMISSION, formats.LEAF_AREA_INDEX,10                  formats.FLOWERING_SCORE, formats.STEM_COUNT,11                  formats.CANOPY_HEIGHT]12    _STAGE_TWO = [formats.DW_PLANT]13    _DELTA_INIT_DATA = formats.delta(_INIT_DATA) +\14        [formats.ROW, formats.COL, formats.GENOTYPE]15    _DELTA_STAGE_ONE = formats.delta(_STAGE_ONE)16    _DELTA_STAGE_TWO = formats.delta(_STAGE_TWO)17    _DELTA_CUMUL_INIT_DATA = list(set(_INIT_DATA + _DELTA_INIT_DATA))18    _DELTA_CUMUL_STAGE_ONE = list(set(_STAGE_ONE + _DELTA_STAGE_ONE))19    _DELTA_CUMUL_STAGE_TWO = list(set(_STAGE_TWO + _DELTA_STAGE_TWO))20    _NAME = "GAWinModel"21    def __init__(self, data):22        """23        data - {'stage_one': {'train_data': [], 'test_data': []},24                'stage_two': {'train_data': [], 'test_data': []}}25        """26        self._data = data27        self._instructions = self._build_instructions()28        self._predictions = self.predict()29    def predict(self, compare=False):30        ####################################################################31        self._submodels = []32        for instruction in self._instructions:33            if instruction['stage'] == 1:34                stage_data = self._data['stage_one']35            elif instruction['stage'] == 2:36                stage_data = self._data['stage_two']37            else:38                raise Exception("Unknown stage %d" % instruction['stage'])39            self._submodels.append(MLModelMemfix(stage_data, instruction))40        ####################################################################41        # extract the INIT variables from the test dataset42        input_data = []43        for entry in self._data['stage_one']['test_data']:44            new_entry = dict()45            # using set because pheno_date may be present in both lists46            for var in set(self._init_vars + self._META_DATA):47                new_entry[var] = entry[var]48            input_data.append(new_entry)49        # go through each stage one model and predict stage two variables50        stage_one_models = self._get_models(1)51        for model in stage_one_models:52            var_name = model.get_name()53            predictions = model.predict(record_real=False, data=input_data)54            # update each input_data entry with the predicted variable55            for prediction in predictions:56                id_keys = self._META_DATA57                if formats.PSEUDO_REP not in prediction.keys():58                    id_keys.remove(formats.PSEUDO_REP)59                filter_me = input_data60                for key in id_keys:61                    filter_me = [x for x in filter_me if62                                 x[key] == prediction[key]]63                # there should be just one entry if everything went well64                assert(len(filter_me) == 1)65                entry = filter_me[0]66                # now update the entry with the predicted variable67                entry[var_name] = prediction['predicted']68        required_keys = self._init_vars + self._pheno_keys69        for entry in input_data:70            for key in required_keys:71                assert(key in entry.keys())72        # get the dry weight prediction model73        stage_two_models = self._get_models(2)74        assert(len(stage_two_models) == 1)75        stage_two_model = stage_two_models[0]76        # start predicting77        predictions = stage_two_model.predict(record_real=False,78                                              data=input_data)79        if predictions[0]['pheno'] == formats.delta(formats.DW_PLANT):80            predictions = self._convert(predictions)81        else:82            # just because the compare function expects the predictions83            # to have a value for DW_PLANT84            for prediction in predictions:85                prediction[prediction['pheno']] = prediction['predicted']86        if compare:87            return self._compare(predictions)88        else:89            return predictions90    def get_predictions(self, compare=False):91        if compare:92            return self._compare(self._predictions)93        else:94            return self._predictions95    def _convert(self, predictions):96        """Converts delta predictions to cumulative values"""97        results = []98        for entry in self._data['init_entries']:99            if entry[formats.DATE].year != predictions[0][formats.DATE].year:100                continue101            new_entry = dict()102            for key in entry.keys():103                if key in self._META_DATA + self._STAGE_TWO:104                    new_entry[key] = entry[key]105            results.append(new_entry)106            if formats.PSEUDO_REP in entry.keys():107                matching = [x for x in predictions if108                            x[formats.UID] == entry[formats.UID] and109                            x[formats.PSEUDO_REP] == entry[formats.PSEUDO_REP]]110            else:111                matching = [x for x in predictions if112                            x[formats.UID] == entry[formats.UID]]113            matching.sort(key=lambda x: x[formats.DATE])114            previous = entry115            for match in matching:116                new_entry = dict()117                # copy meta data118                for key in self._META_DATA:119                    new_entry[key] = match[key]120                new_entry[formats.DW_PLANT] = previous[formats.DW_PLANT] + \121                    match['predicted']122                results.append(new_entry)123                previous = match124                previous[formats.DW_PLANT] = previous['predicted']125        return results126    def _get_models(self, stage):127        return [x for x in self._submodels if x.get_stage() == stage]128    def _compare(self, predictions):129        results = deepcopy(self._data['stage_two']['test_data'])130        for result in results:131            if formats.PSEUDO_REP in result.keys():132                prediction = [x for x in predictions if133                              x[formats.UID] == result[formats.UID] and134                              x[formats.DATE] == result[formats.DATE] and135                              x[formats.PSEUDO_REP] ==136                              result[formats.PSEUDO_REP]]137            else:138                prediction = [x for x in predictions if139                              x[formats.UID] == result[formats.UID] and140                              x[formats.DATE] == result[formats.DATE]]141            assert(len(prediction) == 1)142            prediction = prediction[0]143            result['predicted'] = prediction[formats.DW_PLANT]144        return results145    def _build_instructions(self):146        available_keys = self._data['stage_one']['train_data'][0].keys()147        self._init_vars = [x for x in self._INIT_DATA if x in available_keys]148        # pheno_keys are all pheno_keys that the GAWinModel can handle149        all_pheno_keys = [formats.LEAF_AREA_INDEX, formats.STEM_COUNT,150                          formats.TRANSMISSION, formats.FLOWERING_SCORE]151        self._pheno_keys = []152        instructions = []153        for pheno_key in all_pheno_keys:154            if pheno_key in available_keys:155                # predict: False - turn off prediction of156                # test data during training157                instruction = {'name': pheno_key,158                               'variables': self._init_vars +159                               self._pheno_keys,160                               'stage': 1,161                               'predict': False}162                if pheno_key == formats.LEAF_AREA_INDEX:163                    instruction['method'] = MLModelMemfix._LINEAR_REGRESSION164                elif pheno_key in [formats.STEM_COUNT, formats.TRANSMISSION]:165                    instruction['method'] = MLModelMemfix._SVM166                elif pheno_key == formats.FLOWERING_SCORE:167                    instruction['method'] = MLModelMemfix._KNN168                instructions.append(instruction)169                self._pheno_keys.append(pheno_key)170        instructions.append({'name': formats.DW_PLANT,171                             'method': MLModelMemfix._KNN,172                             'variables': self._init_vars + self._pheno_keys,173                             'predict': False,174                             'stage': 2})...

docker_registry_contact.py

Source:docker_registry_contact.py

...72        self._port = port73        self._url_prefix = 'http://{0}:{1}'.format(self._host, self._port)74    def get_last_error(self):75        return self._last_error76    def _init_vars(self):77        self._last_error = None78    def version_check(self):79        self._init_vars()80        pass81    def list_repositories(self):82        """83            84        :return: [<repo_name>, ...] or None for error occurred85        """86        self._init_vars()87        repositories = None88        url = self._url_prefix + DockerRegistryAPIDisc.api_disc_list_repositories()['uri']89        try:90            r = requests.get(url=url)91            content = r.content.encode('utf-8')92            content = dict(json.loads(content))93            if r.status_code != 200:94                self._last_error = content['errors']95            else:96                repositories = content['repositories']97        except Exception as exp:98            self._last_error = exp99        return repositories100    def list_image_tags(self, image_name):101        """102        :param image_name: 103        104        :return: [<tag>, ...] or None for error occurred105        """106        self._init_vars()107        tags = None108        url = self._url_prefix + DockerRegistryAPIDisc.api_disc_list_image_tags()['uri']109        url = url.format(name=image_name)110        try:111            r = requests.get(url=url)112            content = r.content.encode('utf-8')113            content = dict(json.loads(content))114            if r.status_code != 200:115                self._last_error = content['errors']116            else:117                tags = content['tags']118        except Exception as exp:119            self._last_error = exp120        return tags

wslite.py

Source:wslite.py

...6class Main( xbmc.Monitor ):7    def __init__( self ):8        """Starts the background process for weather station lite."""9        xbmc.Monitor.__init__( self )10        self._init_vars()11        self.LW.log( ['background monitor version %s started' % self.SETTINGS['ADDONVERSION']], xbmc.LOGINFO )12        self.LW.log( ['debug logging set to %s' % self.SETTINGS['debug']], xbmc.LOGINFO )13        self.LW.log( ['starting external hardware interface'] )14        while not self.abortRequested():15            if self.waitForAbort( 10 ):16                break17        self.LW.log( ['background monitor version %s stopped' % self.SETTINGS['ADDONVERSION']], xbmc.LOGINFO )18    def onNotification( self, sender, method, data ):19        if 'Other.RPIWSL_VariablePass' in method:20            data = json.loads( data )21            self.LW.log( ['MONITOR METHOD: %s DATA: %s' % (str( method ), str( data ))] )22            if data == 'ScreenStatus:ScreenOff':23                self._set_property( 'ScreenStatus', 'Off' )24            else:25                self._set_properties( self.SENSORINTERFACE.getSensorData( data=data ) )26        elif 'Other.RPIWSL_SettingsRequest' in method:27            data = json.loads( data )28            self.LW.log( ['MONITOR METHOD: %s DATA: %s' % (str( method ), str( data ))] )29            self.SENSORINTERFACE.handlePassback( 'ReturningSettings', self._scrub_settings() )30    def onSettingsChanged( self ):31        self._init_vars()32        self.SENSORINTERFACE.handlePassback( 'ReturningSettings', self._scrub_settings() )33    def _init_vars( self ):34        self.WINDOWID = 1260035        self.WINDOW = xbmcgui.Window( int(self.WINDOWID) )36        self.SETTINGS = loadSettings()37        self.LW = Logger( preamble='[WS Lite Monitor]', logdebug=self.SETTINGS['debug'] )38        self.SENSORINTERFACE = self._pick_sensor_interface()39    def _pick_sensor_interface( self ):40        if self.SETTINGS['sensor_interface'] == 0:41            return rpiwsl.SensorInterface()42    def _scrub_settings( self ):43        scrubbed_settings = {}44        for thekey, thevalue in self.SETTINGS.items():45            if 'ADDON' in thekey:46                continue47            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.