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

How to use after_init method in molecule

Best Python code snippet using molecule_python

enum.py

Source:enum.py

1"""2Wraps enumerations3"""4from typehandlers import inttype5from typehandlers.base import return_type_matcher, param_type_matcher6from cppclass import CppClass7class Enum(object):8    """9    Class that adds support for a C/C++ enum type10    """11    def __init__(self, name, values, values_prefix='', cpp_namespace=None, outer_class=None,12                 import_from_module=None):13        """14        Creates a new enum wrapper, which should be added to a module with module.add_enum().15        :param name: C name of the enum type16        :param values: a list of strings with all enumeration value names, or list of (name, C-value-expr) tuples.17        :param values_prefix: prefix to add to value names, or None18        :param cpp_namespace: optional C++ namespace identifier, or None.19                         Note: this namespace is *in addition to*20                         whatever namespace of the module the enum21                         belongs to.  Typically this parameter is to22                         be used when wrapping enums declared inside23                         C++ classes.24        :param import_from_module: if not None, the enum is defined in25            another module, this parameter gives the name of the module26        """27        assert isinstance(name, basestring)28        assert '::' not in name29        assert outer_class is None or isinstance(outer_class, CppClass)30        self.outer_class = outer_class31        for val in values:32            if not isinstance(val, (basestring, tuple)):33                raise TypeError34        #if not name:35        #    raise ValueError36        self.name = name37        self.full_name = None38        self.values = list(values)39        self.values_prefix = values_prefix40        self.cpp_namespace = cpp_namespace41        self._module = None42        self.ThisEnumParameter = None43        self.ThisEnumReturn = None44        self.import_from_module = import_from_module45    def get_module(self):46        """Get the Module object this class belongs to"""47        return self._module48    def set_module(self, module):49        """Set the Module object this class belongs to; can only be set once"""50        assert self._module is None51        self._module = module52        if not self.name:53            return54        if self.outer_class is None:55            if self._module.cpp_namespace_prefix:56                if self._module.cpp_namespace_prefix == '::':57                    self.full_name = '::' + self.name58                else:59                    self.full_name = self._module.cpp_namespace_prefix + '::' + self.name60            else:61                self.full_name = self.name62        else:63            self.full_name = '::'.join([self.outer_class.full_name, self.name])64        ## Register type handlers for the enum type65        assert self.name66        assert self.full_name67        class ThisEnumParameter(inttype.IntParam):68            CTYPES = []69            full_type_name = self.full_name70            def __init__(self, ctype, name, *args, **kwargs):71                super(ThisEnumParameter, self).__init__(self.full_type_name, name, *args, **kwargs)72        class ThisEnumReturn(inttype.IntReturn):73            CTYPES = []74            full_type_name = self.full_name75            def __init__(self, ctype, *args, **kwargs):76                super(ThisEnumReturn, self).__init__(self.full_type_name, *args, **kwargs)77        class ThisEnumRefParameter(inttype.IntRefParam):78            CTYPES = []79            full_type_name = self.full_name + " &"80            def __init__(self, ctype, name, *args, **kwargs):81                super(ThisEnumRefParameter, self).__init__(self.full_type_name, name, *args, **kwargs)82        class ThisEnumPtrParameter(inttype.IntPtrParam):83            CTYPES = []84            full_type_name = self.full_name + " *"85            def __init__(self, ctype, name, *args, **kwargs):86                super(ThisEnumPtrParameter, self).__init__(self.full_type_name, name, *args, **kwargs)87        self.ThisEnumParameter = ThisEnumParameter88        self.ThisEnumReturn = ThisEnumReturn89        self.ThisEnumRefParameter = ThisEnumRefParameter90        self.ThisEnumPtrParameter = ThisEnumPtrParameter91        param_type_matcher.register(self.full_name, self.ThisEnumParameter)92        return_type_matcher.register(self.full_name, self.ThisEnumReturn)93        param_type_matcher.register(self.full_name + ' &', self.ThisEnumRefParameter)94        param_type_matcher.register(self.full_name + ' *', self.ThisEnumPtrParameter)95        if self.name != self.full_name:96            try:97                param_type_matcher.register(self.name, self.ThisEnumParameter)98            except ValueError:99                pass100            try:101                return_type_matcher.register(self.name, self.ThisEnumReturn)102            except ValueError:103                pass104            try:105                param_type_matcher.register(self.name+' &', self.ThisEnumRefParameter)106            except ValueError:107                pass108            try:109                param_type_matcher.register(self.name+' *', self.ThisEnumPtrParameter)110            except ValueError:111                pass112    module = property(get_module, set_module)113    def generate(self, unused_code_sink):114        if self.import_from_module:115            return #........ RET116        module = self.module117        if self.outer_class is None:118            namespace = []119            if module.cpp_namespace_prefix:120                namespace.append(module.cpp_namespace_prefix)121            if self.cpp_namespace:122                namespace.append(self.cpp_namespace)123            for value in self.values:124                if isinstance(value, tuple):125                    name, real_value = value126                    module.after_init.write_code(127                        "PyModule_AddIntConstant(m, (char *) \"%s\", %s);" % (name, real_value))128                else:129                    module.after_init.write_code(130                        "PyModule_AddIntConstant(m, (char *) \"%s\", %s);"131                        % (value, '::'.join(namespace + [self.values_prefix + value])))132        else:133            module.after_init.write_code("{")134            module.after_init.indent()135            module.after_init.write_code("PyObject *tmp_value;")136            for value in self.values:137                if isinstance(value, tuple):138                    value_name, value_str = value139                else:140                    value_name = value141                    value_str = "%s::%s" % (self.outer_class.full_name, value)142                module.after_init.write_code(143                    ' // %s\n'144                    'tmp_value = PyInt_FromLong(%s);\n'145                    'PyDict_SetItemString((PyObject*) %s.tp_dict, \"%s\", tmp_value);\n'146                    'Py_DECREF(tmp_value);'147                    % (148                    value_str, value_str, self.outer_class.pytypestruct, value_name))149            module.after_init.unindent()150            module.after_init.write_code("}")151    def generate_declaration(self, sink, module):...

service_base.py

Source:service_base.py

1#!/usr/bin/env python2from functools import partial3import json4import getopt5import logging6import signal7import sys8from common import init_base, init_django, invoke9def _handle_args(argv = sys.argv[1:]):10    config_file = 'config.json'11    try:12        opts, _ = getopt.getopt(13            argv, 'o:fpdiwec',14            ['level=', 'show-warnings', 'output-file=', 'log-format=',      \15                'debug', 'info', 'warn', 'error', 'config-file='])16    except getopt.error, msg:17        print msg18        sys.exit(2)19    logging.basicConfig(20        level = logging.DEBUG)21    now_handler = logging.getLogger().handlers[0]22    # process options23    for o, a in opts:24        if o in ('-o', '--output-file'):25            logging.getLogger('').addHandler(now_handler)26            now_handler = logging.FileHandler(a)27        elif o in ('-l', '--level'):28            now_handler.setLevel(a)29        elif o in ('-d', '--debug'):30            now_handler.setLevel(logging.DEBUG)31        elif o in ('-i', '--info'):32            now_handler.setLevel(logging.INFO)33        elif o in ('-w', '--warn'):34            now_handler.setLevel(logging.WARN)35        elif o in ('-e', '--error'):36            now_handler.setLevel(logging.ERROR)37        elif o in ('-f', '--log-format'):38            now_handler.setFormatter(logging.Formatter(fmt = a))39        elif o in ('-c', '--config-file'):40            config_file = a41        elif o in ('--show-warnings', ):42            logging.captureWarnings(True)43    logging.getLogger('').addHandler(now_handler)44    return config_file45class ServiceEngine(object):46    def __init__(self, local_name):47        from connection import SessionManager48        self.servers     = []49        self.session_mgr = SessionManager()50        self.local_name  = local_name51    def add_server(self, ctor, kwargs, delayed):52        kwargs = kwargs or {}53        kwargs['local_name']  = self.local_name54        kwargs['session_mgr'] = self.session_mgr55        server = ctor(**kwargs)56        server.delayed_start = delayed57        self.servers.append(server)58    def add_connections(self, ctor, kwargs, after_init):59        kwargs = kwargs or {}60        kwargs['mgr']        = self.session_mgr61        kwargs['local_name'] = self.local_name62        client = ctor(**kwargs)63        client.init_session(64            handler = partial(65                self.__after_session_init,66                session    = client,67                after_init = after_init))68    def __after_session_init(self, session, succeed, after_init):69        self.session_mgr.add_managed_session(session = session)70        invoke(after_init, kwargs = {'session': session})71    def start(self):72        for server in self.servers:73            if not server.delayed_start:74                server.start()75            else:76                from tornado.ioloop import IOLoop77                ioloop = IOLoop.instance()78                ioloop.add_callback(callback = server.start)79    def stop(self):80        for server in self.servers:81            server.shutdown()82class ConfigParser(object):83    def __init__(self, file_name):84        self.engine       = None85        self.file_name    = file_name86        self.server_types = {}87        self.client_types = {}88    def get_engine(self):89        return self.engine90    @staticmethod91    def load_module_item(package, name):92        __import__(package)93        module = sys.modules[package]94        return getattr(module, name)95    def handle_base(self, config):96        local_name = config['local_name']97        self.engine = ServiceEngine(local_name)98        queue_thread = config.get('task_threads')99        require_django = bool(config.get('require_django'))100        if queue_thread:101            from common.task import TaskQueue102            task_queue = TaskQueue(num_thread = queue_thread)103            task_queue.make_default()104            task_queue.start()105        if require_django:106            init_django()107    def handle_type(self, config, mapping):108        if not config:109            return110        for entry in config:111            package = entry['package']112            name    = entry['name']113            clsname = entry['class']114            cls     = self.load_module_item(package, clsname)115            mapping[name] = cls116    def handle_message(self, config):117        for entry in config:118            package = entry119            func    = self.load_module_item(package, 'register_message')120            func()121    def handle_endpoint(self, config):122        for entry in config:123            name    = entry['type']124            kwargs  = entry['kwargs']125            enabled = entry['enabled']126            delayed = bool(entry.get('delay_start'))127            if enabled == 'true':128                ctor    = self.server_types[name]129                self.engine.add_server(ctor, kwargs, delayed)130    def handle_connections(self, config):131        if not config:132            return133        for entry in config:134            name    = entry['type']135            kwargs  = entry['kwargs']136            ctor    = self.client_types[name]137            after_init = entry.get('initialize')138            if after_init:139                after_init = self.load_module_item(140                    after_init['package'], after_init['method'])141            self.engine.add_connections(ctor, kwargs, after_init)142    def parse(self):143        try:144            config_str = None145            with open(self.file_name) as f:146                config_str = f.read()147            config = json.loads(config_str)148            self.handle_base(config['base'])149            self.handle_type(config['server'], self.server_types)150            self.handle_message(config['message'])151            self.handle_type(config.get('client'), self.client_types)152            self.handle_endpoint(config['endpoint'])153            self.handle_connections(config.get('connections'))154        except Exception:155            logging.exception('parse config')156            return False157        return True158def _handle_config(config_file):159    parser = ConfigParser(config_file)160    if parser.parse():161        return parser.get_engine()162def sig_handler(*_):163    from tornado.ioloop import IOLoop164    ioloop = IOLoop.instance()165    ioloop.stop()166def main():167    init_base()168    config_file = _handle_args()169    engine = _handle_config(config_file)170    if not engine:171        sys.exit(1)172    engine.start()173    from tornado.ioloop import IOLoop174    from common.task import TaskQueue175    ioloop = IOLoop.instance()176    signal.signal(signal.SIGINT, sig_handler)177    try:178        logging.info('starting io_loop...')179        ioloop.start()180    except:181        logging.exception('ioloop')182    engine.stop()183    TaskQueue.get_instance().stop(wait = True)184if __name__ == '__main__':...

copy of brick_image_processing.py

Source:copy of brick_image_processing.py

1# ***********************************************************************2# ***********************************************************************3from brick import *4# ***********************************************************************5# ***********************************************************************6# Base Class for the Image Library7# ***********************************************************************8class t_Brick_ImageLib ( tLWB_Brick ):9  default_Color_On  = wx.BLUE10  default_Color_Off = wx.GREEN11  # ************************************12  # __init__ of ancestor, will call after_init13  # ************************************14  def after_init ( self ):15    self.Caption = 'Image'16    # set the colors of this library17    self.Color_On  = self.default_Color_On18    #self.Color_Off = self.default_Color_Off19    20    # shape is always a block so we can define it here21    self.shape = ogl.OGL_Rectangle ( self, self.Pos )22# ***********************************************************************23# ***********************************************************************24# ***********************************************************************25class tRotate ( t_Brick_ImageLib ):26  Description = """Rotates an Image,27either by the internal GUI control, or from the external signal.28Images of the following types are accepted: BMP, PNG, GIF, ..29The output format of the rotated image is the same as the input format."""30  def after_init (self):31    # Define the input pins32    # Pin-Name, Data_Type, Necessary, Description33    self.Inputs [1] = \34      ['Image In',  TIO_ARRAY, True,35       'Accepts images of the following types:\n'36       '   BMP, PNG, GIF, ...']37    self.Inputs [2] = \38      ['Rotation',  TIO_NUMBER, False,39       'The rotation ineeds not to be present (may come from internal GUI)\n'40       'Rotation is normally in the range of 0 .. 360 degrees\n'41       'Values outside this range will be taken module (360)']42    self.N_Inputs = len ( self.Inputs )43    # Define the output pins44    self.Outputs [1] = \45      ['Image Out',  TIO_ARRAY, 'Generates a rotated image in the same format as the input']46    self.N_Outputs = len ( self.Outputs )47    # Create the shape48    t_Brick_ImageLib.after_init (self)49    self.Caption = 'Rotate'50# ***********************************************************************51# ***********************************************************************52# ***********************************************************************53class tShow( t_Brick_ImageLib ):54  def after_init (self):55    # Define the input pins56    # Pin-Name, Data_Type, Necessary, Description57    self.Inputs [1] = \58      ['Image In',  TIO_ARRAY, True,59       'Accepts images of the following types:\n'60       '   BMP, PNG, GIF, ...']61    self.N_Inputs = len ( self.Inputs )62    # Create the shape63    t_Brick_ImageLib.after_init (self)64    self.Caption = 'Show'...

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.