def execute(self, operation, *args, **kwargs):
def pop_models():<EOL><INDENT>models = kwargs.pop('<STR_LIT>', None)<EOL>if models is None:<EOL><INDENT>return None<EOL><DEDENT>else:<EOL><INDENT>if isinstance(models, str):<EOL><INDENT>return [models]<EOL><DEDENT>else:<EOL><INDENT>return models<EOL><DEDENT><DEDENT><DEDENT>def check_models(models):<EOL><INDENT>missing_models = set(models) - set(self.models_loaded)<EOL>if missing_models:<EOL><INDENT>raise ModelMissing('<STR_LIT>''<STR_LIT>'.format(str(list(missing_models))[<NUM_LIT:1>:-<NUM_LIT:1>],<EOL>self))<EOL><DEDENT><DEDENT>def build_filter(models, roots):<EOL><INDENT>if '<STR_LIT>' in kwargs:<EOL><INDENT>logger.warning("<STR_LIT>"<EOL>"<STR_LIT>")<EOL><DEDENT>if isinstance(models, str):<EOL><INDENT>models = [models]<EOL><DEDENT>check_models(models)<EOL>filter_ele = etree.Element(filter_tag, type='<STR_LIT>')<EOL>for root in roots:<EOL><INDENT>etree.SubElement(filter_ele, root)<EOL><DEDENT>filter_xml = etree.tostring(filter_ele,<EOL>encoding='<STR_LIT>',<EOL>pretty_print=False)<EOL>logger.debug("<STR_LIT>".format(filter_xml))<EOL>return filter_ele<EOL><DEDENT>def get_access_type(model_name, root):<EOL><INDENT>check_models([model_name])<EOL>node = list(self.models[model_name].tree.iterchildren(tag=root))[<NUM_LIT:0>]<EOL>return node.get('<STR_LIT>')<EOL><DEDENT>if type(operation) is str:<EOL><INDENT>try:<EOL><INDENT>cls = manager.OPERATIONS[operation]<EOL><DEDENT>except KeyError:<EOL><INDENT>supported_operations = list(manager.OPERATIONS.keys())<EOL>raise ValueError("<STR_LIT>".format(str(supported_operations)[<NUM_LIT:1>:-<NUM_LIT:1>],<EOL>operation))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>cls = operation<EOL><DEDENT>if cls == operations.retrieve.Get:<EOL><INDENT>models = pop_models()<EOL>if models is not None:<EOL><INDENT>check_models(models)<EOL>roots = [k for k, v in self.roots.items()<EOL>if v in models and<EOL>(get_access_type(v, k) == '<STR_LIT>' or<EOL>get_access_type(v, k) == '<STR_LIT>')]<EOL>if not roots:<EOL><INDENT>raise ValueError('<STR_LIT>''<STR_LIT>'.format(str(models)[<NUM_LIT:1>:-<NUM_LIT:1>]))<EOL><DEDENT>kwargs['<STR_LIT>'] = build_filter(models, roots)<EOL><DEDENT><DEDENT>elif cls == operations.retrieve.GetConfig:<EOL><INDENT>if not args and '<STR_LIT:source>' not in kwargs:<EOL><INDENT>args = tuple(['<STR_LIT>'])<EOL><DEDENT>models = pop_models()<EOL>if models is not None:<EOL><INDENT>check_models(models)<EOL>roots = [k for k, v in self.roots.items()<EOL>if v in models and<EOL>get_access_type(v, k) == '<STR_LIT>']<EOL>if not roots:<EOL><INDENT>raise ValueError('<STR_LIT>''<STR_LIT>'.format(str(models)[<NUM_LIT:1>:-<NUM_LIT:1>]))<EOL><DEDENT>kwargs['<STR_LIT>'] = build_filter(models, roots)<EOL><DEDENT><DEDENT>elif cls == operations.edit.EditConfig:<EOL><INDENT>if args and isinstance(args[<NUM_LIT:0>], Config):<EOL><INDENT>args_list = list(args)<EOL>args_list[<NUM_LIT:0>] = args[<NUM_LIT:0>].ele<EOL>args = tuple(args_list)<EOL><DEDENT>if '<STR_LIT:target>' not in kwargs and'<STR_LIT>' not inself.server_capabilities and'<STR_LIT>' inself.server_capabilities:<EOL><INDENT>kwargs['<STR_LIT:target>'] = '<STR_LIT>'<EOL><DEDENT><DEDENT>reply = super().execute(cls, *args, **kwargs)<EOL>if isinstance(reply, operations.rpc.RPCReply):<EOL><INDENT>reply.ns = self._get_ns(reply._root)<EOL><DEDENT>if getattr(transport, '<STR_LIT>', None) andisinstance(reply, transport.notify.Notification):<EOL><INDENT>reply.ns = self._get_ns(reply._root_ele)<EOL><DEDENT>return reply<EOL>
execute

        High-level api: Supported operations are get, get_config, get_schema,
        dispatch, edit_config, copy_config, validate, commit, discard_changes,
        delete_config, lock, unlock, close_session, kill_session,
        poweroff_machine and reboot_machine. Since ModelDevice is a subclass of
        manager in ncclient package, any method supported by ncclient is
        available here. Refer to ncclient document for more details.