def _start(self, spin):
for args, kwargs in self.subscribers:
self.subscribers_init.append(rospy.Subscriber(*args, **kwargs))
is_func = isinstance(self.cl, types.FunctionType)
is_class = isinstance(self.cl, types.TypeType)
if is_class:
targ = self.__start_class
elif is_func:
targ = self.__start_func
self.thread = threading.Thread(target=targ,
args=(self.cl,) + self.cl_args,
kwargs=self.cl_kwargs)
self.thread.daemon = True
self.thread.start()
if spin:
rospy.spin()
return self
python类FunctionType()的实例源码
def addMsgHandler(self, address, callback):
"""Register a handler for an OSC-address
- 'address' is the OSC address-string.
the address-string should start with '/' and may not contain '*'
- 'callback' is the function called for incoming OSCMessages that match 'address'.
The callback-function will be called with the same arguments as the 'msgPrinter_handler' below
"""
for chk in '*?,[]{}# ':
if chk in address:
raise OSCServerError("OSC-address string may not contain any characters in '*?,[]{}# '")
if type(callback) not in (types.FunctionType, types.MethodType):
raise OSCServerError("Message callback '%s' is not callable" % repr(callback))
if address != 'default':
address = '/' + address.strip('/')
self.callbacks[address] = callback
def addMsgHandler(self, address, callback):
"""Register a handler for an OSC-address
- 'address' is the OSC address-string.
the address-string should start with '/' and may not contain '*'
- 'callback' is the function called for incoming OSCMessages that match 'address'.
The callback-function will be called with the same arguments as the 'msgPrinter_handler' below
"""
for chk in '*?,[]{}# ':
if chk in address:
raise OSCServerError("OSC-address string may not contain any characters in '*?,[]{}# '")
if type(callback) not in (types.FunctionType, types.MethodType):
raise OSCServerError("Message callback '%s' is not callable" % repr(callback))
if address != 'default':
address = '/' + address.strip('/')
self.callbacks[address] = callback
def get_undecorated_callback(self):
""" Return the callback. If the callback is a decorated function, try to
recover the original function. """
func = self.callback
func = getattr(func, '__func__' if py3k else 'im_func', func)
closure_attr = '__closure__' if py3k else 'func_closure'
while hasattr(func, closure_attr) and getattr(func, closure_attr):
attributes = getattr(func, closure_attr)
func = attributes[0].cell_contents
# in case of decorators with multiple arguments
if not isinstance(func, FunctionType):
# pick first FunctionType instance from multiple arguments
func = filter(lambda x: isinstance(x, FunctionType),
map(lambda x: x.cell_contents, attributes))
func = list(func)[0] # py3 support
return func
def function_named(fn, name):
"""Return a function with a given __name__.
Will assign to __name__ and return the original function if possible on
the Python implementation, otherwise a new function will be constructed.
This function should be phased out as much as possible
in favor of @decorator. Tests that "generate" many named tests
should be modernized.
"""
try:
fn.__name__ = name
except TypeError:
fn = types.FunctionType(fn.__code__, fn.__globals__, name,
fn.__defaults__, fn.__closure__)
return fn
def get_functions(cls, ins=None):
items = list(cls.__dict__.items())
base = cls.__bases__[0]
if base is not Node:
items.extend(base.__dict__.items())
if ins is not None:
cls = ins
funcs = [cls.func]
for k, cur_func in items:
if isinstance(cur_func, types.FunctionType):
if k in ["__init__", "func"]:
continue
cur_func = getattr(cls, k)
if cur_func.__annotations__ != {}:
funcs.append(cur_func)
elif hasattr(cur_func, "is_func") and cur_func.is_func:
funcs.append(cur_func)
if funcs == [cls.func] and cls.func is Node.func:
print(cls, "No funcs?")
return funcs
def validate_error_func(self):
if self.error_func:
if isinstance(self.error_func,types.FunctionType):
ismethod = 0
elif isinstance(self.error_func, types.MethodType):
ismethod = 1
else:
self.log.error("'p_error' defined, but is not a function or method")
self.error = 1
return
eline = func_code(self.error_func).co_firstlineno
efile = func_code(self.error_func).co_filename
self.files[efile] = 1
if (func_code(self.error_func).co_argcount != 1+ismethod):
self.log.error("%s:%d: p_error() requires 1 argument",efile,eline)
self.error = 1
# Get the tokens map
def typeof(self, cdecl):
"""Parse the C type given as a string and return the
corresponding <ctype> object.
It can also be used on 'cdata' instance to get its C type.
"""
if isinstance(cdecl, basestring):
return self._typeof(cdecl)
if isinstance(cdecl, self.CData):
return self._backend.typeof(cdecl)
if isinstance(cdecl, types.BuiltinFunctionType):
res = _builtin_function_type(cdecl)
if res is not None:
return res
if (isinstance(cdecl, types.FunctionType)
and hasattr(cdecl, '_cffi_base_type')):
with self._lock:
return self._get_cached_btype(cdecl._cffi_base_type)
raise TypeError(type(cdecl))
def validate_error_func(self):
if self.error_func:
if isinstance(self.error_func,types.FunctionType):
ismethod = 0
elif isinstance(self.error_func, types.MethodType):
ismethod = 1
else:
self.log.error("'p_error' defined, but is not a function or method")
self.error = 1
return
eline = func_code(self.error_func).co_firstlineno
efile = func_code(self.error_func).co_filename
self.files[efile] = 1
if (func_code(self.error_func).co_argcount != 1+ismethod):
self.log.error("%s:%d: p_error() requires 1 argument",efile,eline)
self.error = 1
# Get the tokens map
def set_action(self, action, *args):
"""Set action and arguments to run in triggered time
:param action: function path as anction
:type action: str
:param args: function arguments
:type args: list or tuple
"""
if isinstance(action, (types.FunctionType, types.MethodType)):
self.action = action
else:
action_module = '.'.join(action.split('.')[:-1])
action_module = importlib.import_module(action_module)
action = action.split('.')[-1]
self.action = getattr(action_module, action)
self.action_args = args
def getFunctions(self):
classes = []
functions = []
try:
blacklist = BLACKLIST[self.module_name]
except KeyError:
blacklist = set()
for name in dir(self.module):
attr = getattr(self.module, name)
if name in blacklist:
continue
if isinstance(attr, (FunctionType, BuiltinFunctionType)):
functions.append(name)
elif isinstance(attr, type):
classes.append(name)
return functions, classes
def __init__(self, inputs=None, include_paths=None, defines=None, library_paths=None, libraries=None):
"""
:param inputs: input paths; note that these should be *absolute* paths
:type inputs: [:obj:`basestring` or :obj:`~types.FunctionType`]
:param include_paths: include paths; note that these should be *absolute* paths
:type include_paths: [:obj:`basestring` or :obj:`~types.FunctionType`]
:param defines: defines in a (name, value) tuple format; use None for value if the define
does not have a value
:type defines: [(:obj:`basestring` or :obj:`~types.FunctionType`, :obj:`basestring` or
:obj:`~types.FunctionType`)]
:param library_paths: include paths; note that these should be *absolute* paths
:type library_paths: [:obj:`basestring` or :obj:`~types.FunctionType`]
:param libraries: library names
:type libraries: [:obj:`basestring` or :obj:`~types.FunctionType`]
"""
super(ExplicitExtension, self).__init__()
self.inputs = StrictList(inputs, value_type=(basestring, FunctionType))
self.include_paths = StrictList(include_paths, value_type=(basestring, FunctionType))
self.defines = defines or []
self.library_paths = StrictList(library_paths, value_type=(basestring, FunctionType))
self.libraries = StrictList(libraries, value_type=(basestring, FunctionType))
def _get_runtime_args(config, workflow_module, workflow_options, current_env):
"""??????????args???????env??args????"""
# ??????args??
args = getattr(workflow_module, "args", {})
if isinstance(args, types.FunctionType):
args = args(workflow_options)
if args is None:
args = {}
# ????????args??
env_args = {} if current_env.args is None else current_env.args
if isinstance(env_args, types.FunctionType):
env_args = env_args(workflow_options)
if env_args is None:
env_args = {}
args.update(env_args)
return args
def __init__(self, module, config=None, options=None):
super(self, ModuleWorkflowBuilder).__init__()
# ?????????
self._clazz = getattr(module, "workflow_class", Workflow)
self._units = getattr(module, "workflow")
self._plugin_mgr = getattr(module, "plugin_manager", plugin_mgr)
self._options = options if options else ObjDictModel()
self._config = config if config is not None else Config()
module_config = getattr(module, "config", None)
# ????????????????
if module_config is not None:
if isinstance(module_config, types.FunctionType):
module_config = module_config(options)
self._config.update(module_config)
self._logger = getattr(module, "logger", None)
self._listeners = getattr(module, "listeners", tuple()) # ???????
def build(self):
# ????????????????options????
if isinstance(self._units, types.FunctionType):
self._units = self._units(self._options)
workflow = self._clazz(
self._units,
config=self._config,
plugin_mgr=self._plugin_mgr,
context_factory=self._context_factory,
logger=self._logger
)
if self._listeners:
for listener in self._listeners:
workflow.add_listener(listener)
return workflow
def latestVersionOf(self, anObject):
"""Get the latest version of an object.
This can handle just about anything callable; instances, functions,
methods, and classes.
"""
t = type(anObject)
if t == types.FunctionType:
return latestFunction(anObject)
elif t == types.MethodType:
if anObject.im_self is None:
return getattr(anObject.im_class, anObject.__name__)
else:
return getattr(anObject.im_self, anObject.__name__)
elif t == types.InstanceType:
# Kick it, if it's out of date.
getattr(anObject, 'nothing', None)
return anObject
elif t == types.ClassType:
return latestClass(anObject)
else:
log.msg('warning returning anObject!')
return anObject
def addMethodNamesToDict(classObj, dict, prefix, baseClass=None):
"""
addMethodNamesToDict(classObj, dict, prefix, baseClass=None) -> dict
this goes through 'classObj' (and its bases) and puts method names
starting with 'prefix' in 'dict' with a value of 1. if baseClass isn't
None, methods will only be added if classObj is-a baseClass
If the class in question has the methods 'prefix_methodname' and
'prefix_methodname2', the resulting dict should look something like:
{"methodname": 1, "methodname2": 1}.
"""
for base in classObj.__bases__:
addMethodNamesToDict(base, dict, prefix, baseClass)
if baseClass is None or baseClass in classObj.__bases__:
for name, method in classObj.__dict__.items():
optName = name[len(prefix):]
if ((type(method) is types.FunctionType)
and (name[:len(prefix)] == prefix)
and (len(optName))):
dict[optName] = 1
def accumulateMethods(obj, dict, prefix='', curClass=None):
"""accumulateMethods(instance, dict, prefix)
I recurse through the bases of instance.__class__, and add methods
beginning with 'prefix' to 'dict', in the form of
{'methodname':*instance*method_object}.
"""
if not curClass:
curClass = obj.__class__
for base in curClass.__bases__:
accumulateMethods(obj, dict, prefix, base)
for name, method in curClass.__dict__.items():
optName = name[len(prefix):]
if ((type(method) is types.FunctionType)
and (name[:len(prefix)] == prefix)
and (len(optName))):
dict[optName] = getattr(obj, name)
def _mk_label(mapper, show_operations, show_attributes, show_datatypes, show_inherited, bordersize):
html = '<<TABLE CELLSPACING="0" CELLPADDING="1" BORDER="0" CELLBORDER="%d" BALIGN="LEFT"><TR><TD><FONT POINT-SIZE="10">%s</FONT></TD></TR>' % (bordersize, mapper.class_.__name__)
def format_col(col):
colstr = '+%s' % (col.name)
if show_datatypes:
colstr += ' : %s' % (col.type.__class__.__name__)
return colstr
if show_attributes:
if not show_inherited:
cols = [c for c in mapper.columns if c.table == mapper.tables[0]]
else:
cols = mapper.columns
html += '<TR><TD ALIGN="LEFT">%s</TD></TR>' % '<BR ALIGN="LEFT"/>'.join(format_col(col) for col in cols)
else:
[format_col(col) for col in sorted(mapper.columns, key=lambda col:not col.primary_key)]
if show_operations:
html += '<TR><TD ALIGN="LEFT">%s</TD></TR>' % '<BR ALIGN="LEFT"/>'.join(
'%s(%s)' % (name,", ".join(default is _mk_label and ("%s") % arg or ("%s=%s" % (arg,repr(default))) for default,arg in
zip((func.__defaults__ and len(func.__code__.co_varnames)-1-(len(func.__defaults__) or 0) or func.__code__.co_argcount-1)*[_mk_label]+list(func.__defaults__ or []), func.__code__.co_varnames[1:])
))
for name,func in list(mapper.class_.__dict__.items()) if isinstance(func, types.FunctionType) and func.__module__ == mapper.class_.__module__
)
html+= '</TABLE>>'
return html
def mark_class(marker):
'''Workaround for https://github.com/pytest-dev/pytest/issues/568'''
import types
def copy_func(f):
try:
return types.FunctionType(f.__code__, f.__globals__,
name=f.__name__, argdefs=f.__defaults__,
closure=f.__closure__)
except AttributeError:
return types.FunctionType(f.func_code, f.func_globals,
name=f.func_name,
argdefs=f.func_defaults,
closure=f.func_closure)
def mark(cls):
for method in dir(cls):
if method.startswith('test_'):
f = copy_func(getattr(cls, method))
setattr(cls, method, marker(f))
return cls
return mark
def __init__(cls, name, bases, dictionary):
if type(cls.__init__) is UnboundMethodType:
initializer = cls.__init__
elif type(cls.__new__) is FunctionType:
initializer = cls.__new__
else:
# noinspection PyUnusedLocal
def initializer(self, *args, **kw): pass
# noinspection PyShadowingNames
@preserve_signature(initializer)
def instance_creator(cls, *args, **kw):
key = (args, tuple(sorted(kw.iteritems())))
try:
hash(key)
except TypeError:
raise TypeError('cannot have singletons for classes with unhashable arguments')
if key not in cls.__instances__:
cls.__instances__[key] = super(Singleton, cls).__call__(*args, **kw)
return cls.__instances__[key]
super(Singleton, cls).__init__(name, bases, dictionary)
cls.__instances__ = {}
cls.__instantiate__ = instance_creator.__get__(cls, type(cls)) # bind instance_creator to cls
def entity(entity_name: str) -> FunctionType:
"""
>>> @entity("test")
... class Test: pass
>>> Test.entity_name
'test'
>>> Test().entity_name
'test'
:param entity_name: str
:return: FunctionType
"""
def decorator(item):
item.entity_name = entity_name
return item
return decorator
def validate_error_func(self):
if self.error_func:
if isinstance(self.error_func,types.FunctionType):
ismethod = 0
elif isinstance(self.error_func, types.MethodType):
ismethod = 1
else:
self.log.error("'p_error' defined, but is not a function or method")
self.error = 1
return
eline = func_code(self.error_func).co_firstlineno
efile = func_code(self.error_func).co_filename
self.files[efile] = 1
if (func_code(self.error_func).co_argcount != 1+ismethod):
self.log.error("%s:%d: p_error() requires 1 argument",efile,eline)
self.error = 1
# Get the tokens map
def get_pfunctions(self):
p_functions = []
for name, item in self.pdict.items():
if name[:2] != 'p_': continue
if name == 'p_error': continue
if isinstance(item,(types.FunctionType,types.MethodType)):
line = func_code(item).co_firstlineno
file = func_code(item).co_filename
p_functions.append((line,file,name,item.__doc__))
# Sort all of the actions by line number
p_functions.sort()
self.pfuncs = p_functions
# Validate all of the p_functions
def typeof(self, cdecl):
"""Parse the C type given as a string and return the
corresponding <ctype> object.
It can also be used on 'cdata' instance to get its C type.
"""
if isinstance(cdecl, basestring):
return self._typeof(cdecl)
if isinstance(cdecl, self.CData):
return self._backend.typeof(cdecl)
if isinstance(cdecl, types.BuiltinFunctionType):
res = _builtin_function_type(cdecl)
if res is not None:
return res
if (isinstance(cdecl, types.FunctionType)
and hasattr(cdecl, '_cffi_base_type')):
with self._lock:
return self._get_cached_btype(cdecl._cffi_base_type)
raise TypeError(type(cdecl))
def _ModifyClass(class_object, testcases, naming_type):
assert not getattr(class_object, '_id_suffix', None), (
'Cannot add parameters to %s,'
' which already has parameterized methods.' % (class_object,))
class_object._id_suffix = id_suffix = {}
# We change the size of __dict__ while we iterate over it,
# which Python 3.x will complain about, so use copy().
for name, obj in class_object.__dict__.copy().items():
if (name.startswith(unittest.TestLoader.testMethodPrefix)
and isinstance(obj, types.FunctionType)):
delattr(class_object, name)
methods = {}
_UpdateClassDictForParamTestCase(
methods, id_suffix, name,
_ParameterizedTestIter(obj, testcases, naming_type))
for name, meth in methods.items():
setattr(class_object, name, meth)
def function_named(fn, name):
"""Return a function with a given __name__.
Will assign to __name__ and return the original function if possible on
the Python implementation, otherwise a new function will be constructed.
This function should be phased out as much as possible
in favor of @decorator. Tests that "generate" many named tests
should be modernized.
"""
try:
fn.__name__ = name
except TypeError:
fn = types.FunctionType(fn.__code__, fn.__globals__, name,
fn.__defaults__, fn.__closure__)
return fn
def _init(init=_init, MdSpi=MdApi, TraderSpi=TraderApi):
import sys
from types import ModuleType, FunctionType as F
f = (lambda f:F(f.__code__,env)) if sys.version_info[0] >= 3 else (lambda f:F(f.func_code,env))
mod = sys.modules[__name__]; Module = type(mod)
if Module is ModuleType:
class Module(ModuleType): pass
mod = Module(__name__); env = mod.__dict__
env.update((k,v) for k,v in globals().items() if k.startswith('__') and k.endswith('__'))
MdSpi = dict((k,f(v)) for k,v in MdSpi.__dict__.items() if k.startswith('On'))
TraderSpi = dict((k,f(v)) for k,v in TraderSpi.__dict__.items() if k.startswith('On'))
sys.modules[__name__] = mod
else:
env = mod.__dict__
for k in ('MdApi','TraderApi','_init'): del env[k]
MdSpi = dict((k,v) for k,v in MdSpi.__dict__.items() if k.startswith('On'))
TraderSpi = dict((k,v) for k,v in TraderSpi.__dict__.items() if k.startswith('On'))
f(init)(Module, MdSpi, TraderSpi)
def copy_func(f, name=None, sinceversion=None, doc=None):
"""
Returns a function with same code, globals, defaults, closure, and
name (or provide a new name).
"""
# See
# http://stackoverflow.com/questions/6527633/how-can-i-make-a-deepcopy-of-a-function-in-python
fn = types.FunctionType(f.__code__, f.__globals__, name or f.__name__, f.__defaults__,
f.__closure__)
# in case f was given attrs (note this dict is a shallow copy):
fn.__dict__.update(f.__dict__)
if doc is not None:
fn.__doc__ = doc
if sinceversion is not None:
fn = since(sinceversion)(fn)
return fn
def sg_inject(path, mod_name):
r"""Converts all functions in the given Python module to sugar functions
so that they can be used in a chainable manner.
Args:
path: A string. Path to the Python module
mod_name: A string. The name of the Python module to inject.
Returns:
None
"""
# import module
import sys
if path not in list(sys.path):
sys.path.append(path)
globals()[mod_name] = importlib.import_module(mod_name)
# find functions
for func_name in dir(globals()[mod_name]):
if isinstance(globals()[mod_name].__dict__.get(func_name), types.FunctionType):
if not func_name.startswith('_'):
# inject to tf.Variable type
exec('tf.Variable.%s = %s.%s' % (func_name, mod_name, func_name))
# inject to tf.Tensor type
exec('tf.Tensor.%s = %s.%s' % (func_name, mod_name, func_name))
