You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2015/04/21 15:47:41 UTC
[05/10] flink git commit: [FLINK-671] Python API
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/objtypes.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/objtypes.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/objtypes.py
new file mode 100644
index 0000000..a6f159e
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/objtypes.py
@@ -0,0 +1,45 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+#!/usr/bin/env python
+#
+# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
+# Copyright (c) 2008-2014 California Institute of Technology.
+# License: 3-clause BSD. The full license text is available at:
+# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
+"""
+all Python Standard Library object types (currently: CH 1-15 @ 2.7)
+and some other common object types (i.e. numpy.ndarray)
+
+to load more objects and types, use dill.load_types()
+"""
+
+from __future__ import absolute_import
+
+# non-local import of dill.objects
+from dill import objects
+for _type in objects.keys():
+ exec("%s = type(objects['%s'])" % (_type,_type))
+
+del objects
+try:
+ del _type
+except NameError:
+ pass
+
+del absolute_import
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/pointers.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/pointers.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/pointers.py
new file mode 100644
index 0000000..2cad3be
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/pointers.py
@@ -0,0 +1,140 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+#!/usr/bin/env python
+#
+# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
+# Copyright (c) 2008-2014 California Institute of Technology.
+# License: 3-clause BSD. The full license text is available at:
+# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
+
+from __future__ import absolute_import
+__all__ = ['parent', 'reference', 'at', 'parents', 'children']
+
+import gc
+import sys
+
+from .dill import _proxy_helper as reference
+from .dill import _locate_object as at
+
+def parent(obj, objtype, ignore=()):
+ """
+>>> listiter = iter([4,5,6,7])
+>>> obj = parent(listiter, list)
+>>> obj == [4,5,6,7] # actually 'is', but don't have handle any longer
+True
+
+NOTE: objtype can be a single type (e.g. int or list) or a tuple of types.
+
+WARNING: if obj is a sequence (e.g. list), may produce unexpected results.
+Parent finds *one* parent (e.g. the last member of the sequence).
+ """
+ depth = 1 #XXX: always looking for the parent (only, right?)
+ chain = parents(obj, objtype, depth, ignore)
+ parent = chain.pop()
+ if parent is obj:
+ return None
+ return parent
+
+
+def parents(obj, objtype, depth=1, ignore=()): #XXX: objtype=object ?
+ """Find the chain of referents for obj. Chain will end with obj.
+
+ objtype: an object type or tuple of types to search for
+ depth: search depth (e.g. depth=2 is 'grandparents')
+ ignore: an object or tuple of objects to ignore in the search
+ """
+ edge_func = gc.get_referents # looking for refs, not back_refs
+ predicate = lambda x: isinstance(x, objtype) # looking for parent type
+ #if objtype is None: predicate = lambda x: True #XXX: in obj.mro() ?
+ ignore = (ignore,) if not hasattr(ignore, '__len__') else ignore
+ ignore = (id(obj) for obj in ignore)
+ chain = find_chain(obj, predicate, edge_func, depth)[::-1]
+ #XXX: should pop off obj... ?
+ return chain
+
+
+def children(obj, objtype, depth=1, ignore=()): #XXX: objtype=object ?
+ """Find the chain of referrers for obj. Chain will start with obj.
+
+ objtype: an object type or tuple of types to search for
+ depth: search depth (e.g. depth=2 is 'grandchildren')
+ ignore: an object or tuple of objects to ignore in the search
+
+ NOTE: a common thing to ignore is all globals, 'ignore=(globals(),)'
+
+ NOTE: repeated calls may yield different results, as python stores
+ the last value in the special variable '_'; thus, it is often good
+ to execute something to replace '_' (e.g. >>> 1+1).
+ """
+ edge_func = gc.get_referrers # looking for back_refs, not refs
+ predicate = lambda x: isinstance(x, objtype) # looking for child type
+ #if objtype is None: predicate = lambda x: True #XXX: in obj.mro() ?
+ ignore = (ignore,) if not hasattr(ignore, '__len__') else ignore
+ ignore = (id(obj) for obj in ignore)
+ chain = find_chain(obj, predicate, edge_func, depth, ignore)
+ #XXX: should pop off obj... ?
+ return chain
+
+
+# more generic helper function (cut-n-paste from objgraph)
+# Source at http://mg.pov.lt/objgraph/
+# Copyright (c) 2008-2010 Marius Gedminas <ma...@pov.lt>
+# Copyright (c) 2010 Stefano Rivera <st...@rivera.za.net>
+# Released under the MIT licence (see objgraph/objgrah.py)
+
+def find_chain(obj, predicate, edge_func, max_depth=20, extra_ignore=()):
+ queue = [obj]
+ depth = {id(obj): 0}
+ parent = {id(obj): None}
+ ignore = set(extra_ignore)
+ ignore.add(id(extra_ignore))
+ ignore.add(id(queue))
+ ignore.add(id(depth))
+ ignore.add(id(parent))
+ ignore.add(id(ignore))
+ ignore.add(id(sys._getframe())) # this function
+ ignore.add(id(sys._getframe(1))) # find_chain/find_backref_chain, likely
+ gc.collect()
+ while queue:
+ target = queue.pop(0)
+ if predicate(target):
+ chain = [target]
+ while parent[id(target)] is not None:
+ target = parent[id(target)]
+ chain.append(target)
+ return chain
+ tdepth = depth[id(target)]
+ if tdepth < max_depth:
+ referrers = edge_func(target)
+ ignore.add(id(referrers))
+ for source in referrers:
+ if id(source) in ignore:
+ continue
+ if id(source) not in depth:
+ depth[id(source)] = tdepth + 1
+ parent[id(source)] = target
+ queue.append(source)
+ return [obj] # not found
+
+
+# backward compatability
+refobject = at
+
+
+# EOF
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/source.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/source.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/source.py
new file mode 100644
index 0000000..b51c007
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/source.py
@@ -0,0 +1,1028 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# #!/usr/bin/env python
+#
+# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
+# Copyright (c) 2008-2014 California Institute of Technology.
+# License: 3-clause BSD. The full license text is available at:
+# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
+#
+# inspired by inspect.py from Python-2.7.6
+# inspect.py author: 'Ka-Ping Yee <pi...@lfw.org>'
+# inspect.py merged into original dill.source by Mike McKerns 4/13/14
+"""
+Extensions to python's 'inspect' module, which can be used
+to retrieve information from live python objects. The methods
+defined in this module are augmented to facilitate access to
+source code of interactively defined functions and classes,
+as well as provide access to source code for objects defined
+in a file.
+"""
+
+from __future__ import absolute_import
+__all__ = ['findsource', 'getsourcelines', 'getsource', 'indent', 'outdent', \
+ '_wrap', 'dumpsource', 'getname', '_namespace', 'getimport', \
+ '_importable', 'importable','isdynamic', 'isfrommain']
+
+import re
+import linecache
+from tokenize import TokenError
+from inspect import ismodule, isclass, ismethod, isfunction, istraceback
+from inspect import isframe, iscode, getfile, getmodule, getsourcefile
+from inspect import getblock, indentsize, isbuiltin
+from .dill import PY3
+
+def isfrommain(obj):
+ "check if object was built in __main__"
+ module = getmodule(obj)
+ if module and module.__name__ == '__main__':
+ return True
+ return False
+
+
+def isdynamic(obj):
+ "check if object was built in the interpreter"
+ try: file = getfile(obj)
+ except TypeError: file = None
+ if file == '<stdin>' and isfrommain(obj):
+ return True
+ return False
+
+
+def _matchlambda(func, line):
+ """check if lambda object 'func' matches raw line of code 'line'"""
+ from dill.detect import code as getcode
+ from dill.detect import freevars, globalvars, varnames
+ dummy = lambda : '__this_is_a_big_dummy_function__'
+ # process the line (removing leading whitespace, etc)
+ lhs,rhs = line.split('lambda ',1)[-1].split(":", 1) #FIXME: if !1 inputs
+ try: #FIXME: unsafe
+ _ = eval("lambda %s : %s" % (lhs,rhs), globals(),locals())
+ except: _ = dummy
+ # get code objects, for comparison
+ _, code = getcode(_).co_code, getcode(func).co_code
+ # check if func is in closure
+ _f = [line.count(i) for i in freevars(func).keys()]
+ if not _f: # not in closure
+ # check if code matches
+ if _ == code: return True
+ return False
+ # weak check on freevars
+ if not all(_f): return False #XXX: VERY WEAK
+ # weak check on varnames and globalvars
+ _f = varnames(func)
+ _f = [line.count(i) for i in _f[0]+_f[1]]
+ if _f and not all(_f): return False #XXX: VERY WEAK
+ _f = [line.count(i) for i in globalvars(func).keys()]
+ if _f and not all(_f): return False #XXX: VERY WEAK
+ # check if func is a double lambda
+ if (line.count('lambda ') > 1) and (lhs in freevars(func).keys()):
+ _lhs,_rhs = rhs.split('lambda ',1)[-1].split(":",1) #FIXME: if !1 inputs
+ try: #FIXME: unsafe
+ _f = eval("lambda %s : %s" % (_lhs,_rhs), globals(),locals())
+ except: _f = dummy
+ # get code objects, for comparison
+ _, code = getcode(_f).co_code, getcode(func).co_code
+ if len(_) != len(code): return False
+ #NOTE: should be same code same order, but except for 't' and '\x88'
+ _ = set((i,j) for (i,j) in zip(_,code) if i != j)
+ if len(_) != 1: return False #('t','\x88')
+ return True
+ # check indentsize
+ if not indentsize(line): return False #FIXME: is this a good check???
+ # check if code 'pattern' matches
+ #XXX: or pattern match against dis.dis(code)? (or use uncompyle2?)
+ _ = _.split(_[0]) # 't' #XXX: remove matching values if starts the same?
+ _f = code.split(code[0]) # '\x88'
+ #NOTE: should be same code different order, with different first element
+ _ = dict(re.match('([\W\D\S])(.*)', _[i]).groups() for i in range(1,len(_)))
+ _f = dict(re.match('([\W\D\S])(.*)', _f[i]).groups() for i in range(1,len(_f)))
+ if (_.keys() == _f.keys()) and (sorted(_.values()) == sorted(_f.values())):
+ return True
+ return False
+
+
+def findsource(object):
+ """Return the entire source file and starting line number for an object.
+ For interactively-defined objects, the 'file' is the interpreter's history.
+
+ The argument may be a module, class, method, function, traceback, frame,
+ or code object. The source code is returned as a list of all the lines
+ in the file and the line number indexes a line in that list. An IOError
+ is raised if the source code cannot be retrieved, while a TypeError is
+ raised for objects where the source code is unavailable (e.g. builtins)."""
+
+ module = getmodule(object)
+ try: file = getfile(module)
+ except TypeError: file = None
+ # use readline when working in interpreter (i.e. __main__ and not file)
+ if module and module.__name__ == '__main__' and not file:
+ import readline
+ lbuf = readline.get_current_history_length()
+ lines = [readline.get_history_item(i)+'\n' for i in range(1,lbuf)]
+ else:
+ try: # special handling for class instances
+ if not isclass(object) and isclass(type(object)): # __class__
+ file = getfile(module)
+ sourcefile = getsourcefile(module)
+ else: # builtins fail with a TypeError
+ file = getfile(object)
+ sourcefile = getsourcefile(object)
+ except (TypeError, AttributeError): # fail with better error
+ file = getfile(object)
+ sourcefile = getsourcefile(object)
+ if not sourcefile and file[:1] + file[-1:] != '<>':
+ raise IOError('source code not available')
+ file = sourcefile if sourcefile else file
+
+ module = getmodule(object, file)
+ if module:
+ lines = linecache.getlines(file, module.__dict__)
+ else:
+ lines = linecache.getlines(file)
+
+ if not lines:
+ raise IOError('could not get source code')
+
+ #FIXME: all below may fail if exec used (i.e. exec('f = lambda x:x') )
+ if ismodule(object):
+ return lines, 0
+
+ name = pat1 = obj = ''
+ pat2 = r'^(\s*@)'
+# pat1b = r'^(\s*%s\W*=)' % name #FIXME: finds 'f = decorate(f)', not exec
+ if ismethod(object):
+ name = object.__name__
+ if name == '<lambda>': pat1 = r'(.*(?<!\w)lambda(:|\s))'
+ else: pat1 = r'^(\s*def\s)'
+ if PY3: object = object.__func__
+ else: object = object.im_func
+ if isfunction(object):
+ name = object.__name__
+ if name == '<lambda>':
+ pat1 = r'(.*(?<!\w)lambda(:|\s))'
+ obj = object #XXX: better a copy?
+ else: pat1 = r'^(\s*def\s)'
+ if PY3: object = object.__code__
+ else: object = object.func_code
+ if istraceback(object):
+ object = object.tb_frame
+ if isframe(object):
+ object = object.f_code
+ if iscode(object):
+ if not hasattr(object, 'co_firstlineno'):
+ raise IOError('could not find function definition')
+ stdin = object.co_filename == '<stdin>'
+ if stdin:
+ lnum = len(lines) - 1 # can't get lnum easily, so leverage pat
+ if not pat1: pat1 = r'^(\s*def\s)|(.*(?<!\w)lambda(:|\s))|^(\s*@)'
+ else:
+ lnum = object.co_firstlineno - 1
+ pat1 = r'^(\s*def\s)|(.*(?<!\w)lambda(:|\s))|^(\s*@)'
+ pat1 = re.compile(pat1); pat2 = re.compile(pat2)
+ #XXX: candidate_lnum = [n for n in range(lnum) if pat1.match(lines[n])]
+ while lnum > 0: #XXX: won't find decorators in <stdin> ?
+ line = lines[lnum]
+ if pat1.match(line):
+ if not stdin: break # co_firstlineno does the job
+ if name == '<lambda>': # hackery needed to confirm a match
+ if _matchlambda(obj, line): break
+ else: # not a lambda, just look for the name
+ if name in line: # need to check for decorator...
+ hats = 0
+ for _lnum in range(lnum-1,-1,-1):
+ if pat2.match(lines[_lnum]): hats += 1
+ else: break
+ lnum = lnum - hats
+ break
+ lnum = lnum - 1
+ return lines, lnum
+
+ try: # turn instances into classes
+ if not isclass(object) and isclass(type(object)): # __class__
+ object = object.__class__ #XXX: sometimes type(class) is better?
+ #XXX: we don't find how the instance was built
+ except AttributeError: pass
+ if isclass(object):
+ name = object.__name__
+ pat = re.compile(r'^(\s*)class\s*' + name + r'\b')
+ # make some effort to find the best matching class definition:
+ # use the one with the least indentation, which is the one
+ # that's most probably not inside a function definition.
+ candidates = []
+ for i in range(len(lines)-1,-1,-1):
+ match = pat.match(lines[i])
+ if match:
+ # if it's at toplevel, it's already the best one
+ if lines[i][0] == 'c':
+ return lines, i
+ # else add whitespace to candidate list
+ candidates.append((match.group(1), i))
+ if candidates:
+ # this will sort by whitespace, and by line number,
+ # less whitespace first #XXX: should sort high lnum before low
+ candidates.sort()
+ return lines, candidates[0][1]
+ else:
+ raise IOError('could not find class definition')
+ raise IOError('could not find code object')
+
+
+def getblocks(object, lstrip=False, enclosing=False, locate=False):
+ """Return a list of source lines and starting line number for an object.
+ Interactively-defined objects refer to lines in the interpreter's history.
+
+ If enclosing=True, then also return any enclosing code.
+ If lstrip=True, ensure there is no indentation in the first line of code.
+ If locate=True, then also return the line number for the block of code.
+
+ DEPRECATED: use 'getsourcelines' instead
+ """
+ lines, lnum = findsource(object)
+
+ if ismodule(object):
+ if lstrip: lines = _outdent(lines)
+ return ([lines], [0]) if locate is True else [lines]
+
+ #XXX: 'enclosing' means: closures only? or classes and files?
+ indent = indentsize(lines[lnum])
+ block = getblock(lines[lnum:]) #XXX: catch any TokenError here?
+
+ if not enclosing or not indent:
+ if lstrip: block = _outdent(block)
+ return ([block], [lnum]) if locate is True else [block]
+
+ pat1 = r'^(\s*def\s)|(.*(?<!\w)lambda(:|\s))'; pat1 = re.compile(pat1)
+ pat2 = r'^(\s*@)'; pat2 = re.compile(pat2)
+ #pat3 = r'^(\s*class\s)'; pat3 = re.compile(pat3) #XXX: enclosing class?
+ #FIXME: bound methods need enclosing class (and then instantiation)
+ # *or* somehow apply a partial using the instance
+
+ skip = 0
+ line = 0
+ blocks = []; _lnum = []
+ target = ''.join(block)
+ while line <= lnum: #XXX: repeat lnum? or until line < lnum?
+ # see if starts with ('def','lambda') and contains our target block
+ if pat1.match(lines[line]):
+ if not skip:
+ try: code = getblock(lines[line:])
+ except TokenError: code = [lines[line]]
+ if indentsize(lines[line]) > indent: #XXX: should be >= ?
+ line += len(code) - skip
+ elif target in ''.join(code):
+ blocks.append(code) # save code block as the potential winner
+ _lnum.append(line - skip) # save the line number for the match
+ line += len(code) - skip
+ else:
+ line += 1
+ skip = 0
+ # find skip: the number of consecutive decorators
+ elif pat2.match(lines[line]):
+ try: code = getblock(lines[line:])
+ except TokenError: code = [lines[line]]
+ skip = 1
+ for _line in code[1:]: # skip lines that are decorators
+ if not pat2.match(_line): break
+ skip += 1
+ line += skip
+ # no match: reset skip and go to the next line
+ else:
+ line +=1
+ skip = 0
+
+ if not blocks:
+ blocks = [block]
+ _lnum = [lnum]
+ if lstrip: blocks = [_outdent(block) for block in blocks]
+ # return last match
+ return (blocks, _lnum) if locate is True else blocks
+
+
+def getsourcelines(object, lstrip=False, enclosing=False):
+ """Return a list of source lines and starting line number for an object.
+ Interactively-defined objects refer to lines in the interpreter's history.
+
+ The argument may be a module, class, method, function, traceback, frame,
+ or code object. The source code is returned as a list of the lines
+ corresponding to the object and the line number indicates where in the
+ original source file the first line of code was found. An IOError is
+ raised if the source code cannot be retrieved, while a TypeError is
+ raised for objects where the source code is unavailable (e.g. builtins).
+
+ If lstrip=True, ensure there is no indentation in the first line of code.
+ If enclosing=True, then also return any enclosing code."""
+ code, n = getblocks(object, lstrip=lstrip, enclosing=enclosing, locate=True)
+ return code[-1], n[-1]
+
+
+#NOTE: broke backward compatibility 4/16/14 (was lstrip=True, force=True)
+def getsource(object, alias='', lstrip=False, enclosing=False, \
+ force=False, builtin=False):
+ """Return the text of the source code for an object. The source code for
+ interactively-defined objects are extracted from the interpreter's history.
+
+ The argument may be a module, class, method, function, traceback, frame,
+ or code object. The source code is returned as a single string. An
+ IOError is raised if the source code cannot be retrieved, while a
+ TypeError is raised for objects where the source code is unavailable
+ (e.g. builtins).
+
+ If alias is provided, then add a line of code that renames the object.
+ If lstrip=True, ensure there is no indentation in the first line of code.
+ If enclosing=True, then also return any enclosing code.
+ If force=True, catch (TypeError,IOError) and try to use import hooks.
+ If builtin=True, force an import for any builtins
+ """
+ # hascode denotes a callable
+ hascode = _hascode(object)
+ # is a class instance type (and not in builtins)
+ instance = _isinstance(object)
+
+ # get source lines; if fail, try to 'force' an import
+ try: # fails for builtins, and other assorted object types
+ lines, lnum = getsourcelines(object, enclosing=enclosing)
+ except (TypeError, IOError): # failed to get source, resort to import hooks
+ if not force: # don't try to get types that findsource can't get
+ raise
+ if not getmodule(object): # get things like 'None' and '1'
+ if not instance: return getimport(object, alias, builtin=builtin)
+ # special handling (numpy arrays, ...)
+ _import = getimport(object, builtin=builtin)
+ name = getname(object, force=True)
+ _alias = "%s = " % alias if alias else ""
+ if alias == name: _alias = ""
+ return _import+_alias+"%s\n" % name
+ else: #FIXME: could use a good bit of cleanup, since using getimport...
+ if not instance: return getimport(object, alias, builtin=builtin)
+ # now we are dealing with an instance...
+ name = object.__class__.__name__
+ module = object.__module__
+ if module in ['builtins','__builtin__']:
+ return getimport(object, alias, builtin=builtin)
+ else: #FIXME: leverage getimport? use 'from module import name'?
+ lines, lnum = ["%s = __import__('%s', fromlist=['%s']).%s\n" % (name,module,name,name)], 0
+ obj = eval(lines[0].lstrip(name + ' = '))
+ lines, lnum = getsourcelines(obj, enclosing=enclosing)
+
+ # strip leading indent (helps ensure can be imported)
+ if lstrip or alias:
+ lines = _outdent(lines)
+
+ # instantiate, if there's a nice repr #XXX: BAD IDEA???
+ if instance: #and force: #XXX: move into findsource or getsourcelines ?
+ if '(' in repr(object): lines.append('%r\n' % object)
+ #else: #XXX: better to somehow to leverage __reduce__ ?
+ # reconstructor,args = object.__reduce__()
+ # _ = reconstructor(*args)
+ else: # fall back to serialization #XXX: bad idea?
+ #XXX: better not duplicate work? #XXX: better new/enclose=True?
+ lines = dumpsource(object, alias='', new=force, enclose=False)
+ lines, lnum = [line+'\n' for line in lines.split('\n')][:-1], 0
+ #else: object.__code__ # raise AttributeError
+
+ # add an alias to the source code
+ if alias:
+ if hascode:
+ skip = 0
+ for line in lines: # skip lines that are decorators
+ if not line.startswith('@'): break
+ skip += 1
+ #XXX: use regex from findsource / getsourcelines ?
+ if lines[skip].lstrip().startswith('def '): # we have a function
+ if alias != object.__name__:
+ lines.append('\n%s = %s\n' % (alias, object.__name__))
+ elif 'lambda ' in lines[skip]: # we have a lambda
+ if alias != lines[skip].split('=')[0].strip():
+ lines[skip] = '%s = %s' % (alias, lines[skip])
+ else: # ...try to use the object's name
+ if alias != object.__name__:
+ lines.append('\n%s = %s\n' % (alias, object.__name__))
+ else: # class or class instance
+ if instance:
+ if alias != lines[-1].split('=')[0].strip():
+ lines[-1] = ('%s = ' % alias) + lines[-1]
+ else:
+ name = getname(object, force=True) or object.__name__
+ if alias != name:
+ lines.append('\n%s = %s\n' % (alias, name))
+ return ''.join(lines)
+
+
+def _hascode(object):
+ '''True if object has an attribute that stores it's __code__'''
+ return getattr(object,'__code__',None) or getattr(object,'func_code',None)
+
+def _isinstance(object):
+ '''True if object is a class instance type (and is not a builtin)'''
+ if _hascode(object) or isclass(object) or ismodule(object):
+ return False
+ if istraceback(object) or isframe(object) or iscode(object):
+ return False
+ # special handling (numpy arrays, ...)
+ if not getmodule(object) and getmodule(type(object)).__name__ in ['numpy']:
+ return True
+# # check if is instance of a builtin
+# if not getmodule(object) and getmodule(type(object)).__name__ in ['__builtin__','builtins']:
+# return False
+ _types = ('<class ',"<type 'instance'>")
+ if not repr(type(object)).startswith(_types): #FIXME: weak hack
+ return False
+ if not getmodule(object) or object.__module__ in ['builtins','__builtin__'] or getname(object, force=True) in ['array']:
+ return False
+ return True # by process of elimination... it's what we want
+
+
+def _intypes(object):
+ '''check if object is in the 'types' module'''
+ import types
+ # allow user to pass in object or object.__name__
+ if type(object) is not type(''):
+ object = getname(object, force=True)
+ if object == 'ellipsis': object = 'EllipsisType'
+ return True if hasattr(types, object) else False
+
+
+def _isstring(object): #XXX: isstringlike better?
+ '''check if object is a string-like type'''
+ if PY3: return isinstance(object, (str, bytes))
+ return isinstance(object, basestring)
+
+
+def indent(code, spaces=4):
+ '''indent a block of code with whitespace (default is 4 spaces)'''
+ indent = indentsize(code)
+ if type(spaces) is int: spaces = ' '*spaces
+ # if '\t' is provided, will indent with a tab
+ nspaces = indentsize(spaces)
+ # blank lines (etc) need to be ignored
+ lines = code.split('\n')
+## stq = "'''"; dtq = '"""'
+## in_stq = in_dtq = False
+ for i in range(len(lines)):
+ #FIXME: works... but shouldn't indent 2nd+ lines of multiline doc
+ _indent = indentsize(lines[i])
+ if indent > _indent: continue
+ lines[i] = spaces+lines[i]
+## #FIXME: may fail when stq and dtq in same line (depends on ordering)
+## nstq, ndtq = lines[i].count(stq), lines[i].count(dtq)
+## if not in_dtq and not in_stq:
+## lines[i] = spaces+lines[i] # we indent
+## # entering a comment block
+## if nstq%2: in_stq = not in_stq
+## if ndtq%2: in_dtq = not in_dtq
+## # leaving a comment block
+## elif in_dtq and ndtq%2: in_dtq = not in_dtq
+## elif in_stq and nstq%2: in_stq = not in_stq
+## else: pass
+ if lines[-1].strip() == '': lines[-1] = ''
+ return '\n'.join(lines)
+
+
+def _outdent(lines, spaces=None, all=True):
+ '''outdent lines of code, accounting for docs and line continuations'''
+ indent = indentsize(lines[0])
+ if spaces is None or spaces > indent or spaces < 0: spaces = indent
+ for i in range(len(lines) if all else 1):
+ #FIXME: works... but shouldn't outdent 2nd+ lines of multiline doc
+ _indent = indentsize(lines[i])
+ if spaces > _indent: _spaces = _indent
+ else: _spaces = spaces
+ lines[i] = lines[i][_spaces:]
+ return lines
+
+def outdent(code, spaces=None, all=True):
+ '''outdent a block of code (default is to strip all leading whitespace)'''
+ indent = indentsize(code)
+ if spaces is None or spaces > indent or spaces < 0: spaces = indent
+ #XXX: will this delete '\n' in some cases?
+ if not all: return code[spaces:]
+ return '\n'.join(_outdent(code.split('\n'), spaces=spaces, all=all))
+
+
+#XXX: not sure what the point of _wrap is...
+#exec_ = lambda s, *a: eval(compile(s, '<string>', 'exec'), *a)
+__globals__ = globals()
+__locals__ = locals()
+wrap2 = '''
+def _wrap(f):
+ """ encapsulate a function and it's __import__ """
+ def func(*args, **kwds):
+ try:
+ #_ = eval(getsource(f, force=True)) #FIXME: safer, but not as robust
+ exec getimportable(f, alias='_') in %s, %s
+ except:
+ raise ImportError('cannot import name ' + f.__name__)
+ return _(*args, **kwds)
+ func.__name__ = f.__name__
+ func.__doc__ = f.__doc__
+ return func
+''' % ('__globals__', '__locals__')
+wrap3 = '''
+def _wrap(f):
+ """ encapsulate a function and it's __import__ """
+ def func(*args, **kwds):
+ try:
+ #_ = eval(getsource(f, force=True)) #FIXME: safer, but not as robust
+ exec(getimportable(f, alias='_'), %s, %s)
+ except:
+ raise ImportError('cannot import name ' + f.__name__)
+ return _(*args, **kwds)
+ func.__name__ = f.__name__
+ func.__doc__ = f.__doc__
+ return func
+''' % ('__globals__', '__locals__')
+if PY3:
+ exec(wrap3)
+else:
+ exec(wrap2)
+del wrap2, wrap3
+
+
+def _enclose(object, alias=''): #FIXME: needs alias to hold returned object
+ """create a function enclosure around the source of some object"""
+ #XXX: dummy and stub should append a random string
+ dummy = '__this_is_a_big_dummy_enclosing_function__'
+ stub = '__this_is_a_stub_variable__'
+ code = 'def %s():\n' % dummy
+ code += indent(getsource(object, alias=stub, lstrip=True, force=True))
+ code += indent('return %s\n' % stub)
+ if alias: code += '%s = ' % alias
+ code += '%s(); del %s\n' % (dummy, dummy)
+ #code += "globals().pop('%s',lambda :None)()\n" % dummy
+ return code
+
+
+def dumpsource(object, alias='', new=False, enclose=True):
+ """'dump to source', where the code includes a pickled object.
+
+ If new=True and object is a class instance, then create a new
+ instance using the unpacked class source code. If enclose, then
+ create the object inside a function enclosure (thus minimizing
+ any global namespace pollution).
+ """
+ from dill import dumps
+ pik = repr(dumps(object))
+ code = 'import dill\n'
+ if enclose:
+ stub = '__this_is_a_stub_variable__' #XXX: *must* be same _enclose.stub
+ pre = '%s = ' % stub
+ new = False #FIXME: new=True doesn't work with enclose=True
+ else:
+ stub = alias
+ pre = '%s = ' % stub if alias else alias
+
+ # if a 'new' instance is not needed, then just dump and load
+ if not new or not _isinstance(object):
+ code += pre + 'dill.loads(%s)\n' % pik
+ else: #XXX: other cases where source code is needed???
+ code += getsource(object.__class__, alias='', lstrip=True, force=True)
+ mod = repr(object.__module__) # should have a module (no builtins here)
+ if PY3:
+ code += pre + 'dill.loads(%s.replace(b%s,bytes(__name__,"UTF-8")))\n' % (pik,mod)
+ else:
+ code += pre + 'dill.loads(%s.replace(%s,__name__))\n' % (pik,mod)
+ #code += 'del %s' % object.__class__.__name__ #NOTE: kills any existing!
+
+ if enclose:
+ # generation of the 'enclosure'
+ dummy = '__this_is_a_big_dummy_object__'
+ dummy = _enclose(dummy, alias=alias)
+ # hack to replace the 'dummy' with the 'real' code
+ dummy = dummy.split('\n')
+ code = dummy[0]+'\n' + indent(code) + '\n'.join(dummy[-3:])
+
+ return code #XXX: better 'dumpsourcelines', returning list of lines?
+
+
+def getname(obj, force=False): #XXX: allow 'throw'(?) to raise error on fail?
+ """get the name of the object. for lambdas, get the name of the pointer """
+ module = getmodule(obj)
+ if not module: # things like "None" and "1"
+ if not force: return None
+ return repr(obj)
+ try:
+ #XXX: 'wrong' for decorators and curried functions ?
+ # if obj.func_closure: ...use logic from getimportable, etc ?
+ name = obj.__name__
+ if name == '<lambda>':
+ return getsource(obj).split('=',1)[0].strip()
+ # handle some special cases
+ if module.__name__ in ['builtins','__builtin__']:
+ if name == 'ellipsis': name = 'EllipsisType'
+ return name
+ except AttributeError: #XXX: better to just throw AttributeError ?
+ if not force: return None
+ name = repr(obj)
+ if name.startswith('<'): # or name.split('('):
+ return None
+ return name
+
+
+def _namespace(obj):
+ """_namespace(obj); return namespace hierarchy (as a list of names)
+ for the given object. For an instance, find the class hierarchy.
+
+ For example:
+
+ >>> from functools import partial
+ >>> p = partial(int, base=2)
+ >>> _namespace(p)
+ [\'functools\', \'partial\']
+ """
+ # mostly for functions and modules and such
+ #FIXME: 'wrong' for decorators and curried functions
+ try: #XXX: needs some work and testing on different types
+ module = qual = str(getmodule(obj)).split()[1].strip('"').strip("'")
+ qual = qual.split('.')
+ if ismodule(obj):
+ return qual
+ # get name of a lambda, function, etc
+ name = getname(obj) or obj.__name__ # failing, raise AttributeError
+ # check special cases (NoneType, ...)
+ if module in ['builtins','__builtin__']: # BuiltinFunctionType
+ if _intypes(name): return ['types'] + [name]
+ return qual + [name] #XXX: can be wrong for some aliased objects
+ except: pass
+ # special case: numpy.inf and numpy.nan (we don't want them as floats)
+ if str(obj) in ['inf','nan','Inf','NaN']: # is more, but are they needed?
+ return ['numpy'] + [str(obj)]
+ # mostly for classes and class instances and such
+ module = getattr(obj.__class__, '__module__', None)
+ qual = str(obj.__class__)
+ try: qual = qual[qual.index("'")+1:-2]
+ except ValueError: pass # str(obj.__class__) made the 'try' unnecessary
+ qual = qual.split(".")
+ if module in ['builtins','__builtin__']:
+ # check special cases (NoneType, Ellipsis, ...)
+ if qual[-1] == 'ellipsis': qual[-1] = 'EllipsisType'
+ if _intypes(qual[-1]): module = 'types' #XXX: BuiltinFunctionType
+ qual = [module] + qual
+ return qual
+
+
+#NOTE: 05/25/14 broke backward compatability: added 'alias' as 3rd argument
+def _getimport(head, tail, alias='', verify=True, builtin=False):
+ """helper to build a likely import string from head and tail of namespace.
+ ('head','tail') are used in the following context: "from head import tail"
+
+ If verify=True, then test the import string before returning it.
+ If builtin=True, then force an import for builtins where possible.
+ If alias is provided, then rename the object on import.
+ """
+ # special handling for a few common types
+ if tail in ['Ellipsis', 'NotImplemented'] and head in ['types']:
+ head = len.__module__
+ elif tail in ['None'] and head in ['types']:
+ _alias = '%s = ' % alias if alias else ''
+ if alias == tail: _alias = ''
+ return _alias+'%s\n' % tail
+ # we don't need to import from builtins, so return ''
+# elif tail in ['NoneType','int','float','long','complex']: return '' #XXX: ?
+ if head in ['builtins','__builtin__']:
+ # special cases (NoneType, Ellipsis, ...) #XXX: BuiltinFunctionType
+ if tail == 'ellipsis': tail = 'EllipsisType'
+ if _intypes(tail): head = 'types'
+ elif not builtin:
+ _alias = '%s = ' % alias if alias else ''
+ if alias == tail: _alias = ''
+ return _alias+'%s\n' % tail
+ else: pass # handle builtins below
+ # get likely import string
+ if not head: _str = "import %s" % tail
+ else: _str = "from %s import %s" % (head, tail)
+ _alias = " as %s\n" % alias if alias else "\n"
+ if alias == tail: _alias = "\n"
+ _str += _alias
+ # FIXME: fails on most decorators, currying, and such...
+ # (could look for magic __wrapped__ or __func__ attr)
+ # (could fix in 'namespace' to check obj for closure)
+ if verify and not head.startswith('dill.'):# weird behavior for dill
+ #print(_str)
+ try: exec(_str) #XXX: check if == obj? (name collision)
+ except ImportError: #XXX: better top-down or bottom-up recursion?
+ _head = head.rsplit(".",1)[0] #(or get all, then compare == obj?)
+ if not _head: raise
+ if _head != head:
+ _str = _getimport(_head, tail, alias, verify)
+ return _str
+
+
+#XXX: rename builtin to force? vice versa? verify to force? (as in getsource)
+#NOTE: 05/25/14 broke backward compatability: added 'alias' as 2nd argument
+def getimport(obj, alias='', verify=True, builtin=False, enclosing=False):
+ """get the likely import string for the given object
+
+ obj is the object to inspect
+ If verify=True, then test the import string before returning it.
+ If builtin=True, then force an import for builtins where possible.
+ If enclosing=True, get the import for the outermost enclosing callable.
+ If alias is provided, then rename the object on import.
+ """
+ if enclosing:
+ from dill.detect import outermost
+ _obj = outermost(obj)
+ obj = _obj if _obj else obj
+ # get the namespace
+ qual = _namespace(obj)
+ head = '.'.join(qual[:-1])
+ tail = qual[-1]
+ # for named things... with a nice repr #XXX: move into _namespace?
+ try: # look for '<...>' and be mindful it might be in lists, dicts, etc...
+ name = repr(obj).split('<',1)[1].split('>',1)[1]
+ name = None # we have a 'object'-style repr
+ except: # it's probably something 'importable'
+ if head in ['builtins','__builtin__']:
+ name = repr(obj) #XXX: catch [1,2], (1,2), set([1,2])... others?
+ else:
+ name = repr(obj).split('(')[0]
+ #if not repr(obj).startswith('<'): name = repr(obj).split('(')[0]
+ #else: name = None
+ if name: # try using name instead of tail
+ try: return _getimport(head, name, alias, verify, builtin)
+ except ImportError: pass
+ except SyntaxError:
+ if head in ['builtins','__builtin__']:
+ _alias = '%s = ' % alias if alias else ''
+ if alias == name: _alias = ''
+ return _alias+'%s\n' % name
+ else: pass
+ try:
+ #if type(obj) is type(abs): _builtin = builtin # BuiltinFunctionType
+ #else: _builtin = False
+ return _getimport(head, tail, alias, verify, builtin)
+ except ImportError:
+ raise # could do some checking against obj
+ except SyntaxError:
+ if head in ['builtins','__builtin__']:
+ _alias = '%s = ' % alias if alias else ''
+ if alias == tail: _alias = ''
+ return _alias+'%s\n' % tail
+ raise # could do some checking against obj
+
+
+def _importable(obj, alias='', source=None, enclosing=False, force=True, \
+ builtin=True, lstrip=True):
+ """get an import string (or the source code) for the given object
+
+ This function will attempt to discover the name of the object, or the repr
+ of the object, or the source code for the object. To attempt to force
+ discovery of the source code, use source=True, to attempt to force the
+ use of an import, use source=False; otherwise an import will be sought
+ for objects not defined in __main__. The intent is to build a string
+ that can be imported from a python file. obj is the object to inspect.
+ If alias is provided, then rename the object with the given alias.
+
+ If source=True, use these options:
+ If enclosing=True, then also return any enclosing code.
+ If force=True, catch (TypeError,IOError) and try to use import hooks.
+ If lstrip=True, ensure there is no indentation in the first line of code.
+
+ If source=False, use these options:
+ If enclosing=True, get the import for the outermost enclosing callable.
+ If force=True, then don't test the import string before returning it.
+ If builtin=True, then force an import for builtins where possible.
+ """
+ if source is None:
+ source = True if isfrommain(obj) else False
+ if source: # first try to get the source
+ try:
+ return getsource(obj, alias, enclosing=enclosing, \
+ force=force, lstrip=lstrip, builtin=builtin)
+ except: pass
+ try:
+ if not _isinstance(obj):
+ return getimport(obj, alias, enclosing=enclosing, \
+ verify=(not force), builtin=builtin)
+ # first 'get the import', then 'get the instance'
+ _import = getimport(obj, enclosing=enclosing, \
+ verify=(not force), builtin=builtin)
+ name = getname(obj, force=True)
+ if not name:
+ raise AttributeError("object has no atribute '__name__'")
+ _alias = "%s = " % alias if alias else ""
+ if alias == name: _alias = ""
+ return _import+_alias+"%s\n" % name
+
+ except: pass
+ if not source: # try getsource, only if it hasn't been tried yet
+ try:
+ return getsource(obj, alias, enclosing=enclosing, \
+ force=force, lstrip=lstrip, builtin=builtin)
+ except: pass
+ # get the name (of functions, lambdas, and classes)
+ # or hope that obj can be built from the __repr__
+ #XXX: what to do about class instances and such?
+ obj = getname(obj, force=force)
+ # we either have __repr__ or __name__ (or None)
+ if not obj or obj.startswith('<'):
+ raise AttributeError("object has no atribute '__name__'")
+ _alias = '%s = ' % alias if alias else ''
+ if alias == obj: _alias = ''
+ return _alias+'%s\n' % obj
+ #XXX: possible failsafe... (for example, for instances when source=False)
+ # "import dill; result = dill.loads(<pickled_object>); # repr(<object>)"
+
+def _closuredimport(func, alias='', builtin=False):
+ """get import for closured objects; return a dict of 'name' and 'import'"""
+ import re
+ from dill.detect import freevars, outermost
+ free_vars = freevars(func)
+ func_vars = {}
+ # split into 'funcs' and 'non-funcs'
+ for name,obj in list(free_vars.items()):
+ if not isfunction(obj): continue
+ # get import for 'funcs'
+ fobj = free_vars.pop(name)
+ src = getsource(fobj)
+ if src.lstrip().startswith('@'): # we have a decorator
+ src = getimport(fobj, alias=alias, builtin=builtin)
+ else: # we have to "hack" a bit... and maybe be lucky
+ encl = outermost(func)
+ # pattern: 'func = enclosing(fobj'
+ pat = '.*[\w\s]=\s*'+getname(encl)+'\('+getname(fobj)
+ mod = getname(getmodule(encl))
+ #HACK: get file containing 'outer' function; is func there?
+ lines,_ = findsource(encl)
+ candidate = [line for line in lines if getname(encl) in line and \
+ re.match(pat, line)]
+ if not candidate:
+ mod = getname(getmodule(fobj))
+ #HACK: get file containing 'inner' function; is func there?
+ lines,_ = findsource(fobj)
+ candidate = [line for line in lines \
+ if getname(fobj) in line and re.match(pat, line)]
+ if not len(candidate): raise TypeError('import could not be found')
+ candidate = candidate[-1]
+ name = candidate.split('=',1)[0].split()[-1].strip()
+ src = _getimport(mod, name, alias=alias, builtin=builtin)
+ func_vars[name] = src
+ if not func_vars:
+ name = outermost(func)
+ mod = getname(getmodule(name))
+ if not mod or name is func: # then it can be handled by getimport
+ name = getname(func, force=True) #XXX: better key?
+ src = getimport(func, alias=alias, builtin=builtin)
+ else:
+ lines,_ = findsource(name)
+ # pattern: 'func = enclosing('
+ candidate = [line for line in lines if getname(name) in line and \
+ re.match('.*[\w\s]=\s*'+getname(name)+'\(', line)]
+ if not len(candidate): raise TypeError('import could not be found')
+ candidate = candidate[-1]
+ name = candidate.split('=',1)[0].split()[-1].strip()
+ src = _getimport(mod, name, alias=alias, builtin=builtin)
+ func_vars[name] = src
+ return func_vars
+
+#XXX: should be able to use __qualname__
+def _closuredsource(func, alias=''):
+ """get source code for closured objects; return a dict of 'name'
+ and 'code blocks'"""
+ #FIXME: this entire function is a messy messy HACK
+ # - pollutes global namespace
+ # - fails if name of freevars are reused
+ # - can unnecessarily duplicate function code
+ from dill.detect import freevars
+ free_vars = freevars(func)
+ func_vars = {}
+ # split into 'funcs' and 'non-funcs'
+ for name,obj in list(free_vars.items()):
+ if not isfunction(obj):
+ # get source for 'non-funcs'
+ free_vars[name] = getsource(obj, force=True, alias=name)
+ continue
+ # get source for 'funcs'
+ fobj = free_vars.pop(name)
+ src = getsource(fobj, alias) # DO NOT include dependencies
+ # if source doesn't start with '@', use name as the alias
+ if not src.lstrip().startswith('@'): #FIXME: 'enclose' in dummy;
+ src = importable(fobj,alias=name)# wrong ref 'name'
+ org = getsource(func, alias, enclosing=False, lstrip=True)
+ src = (src, org) # undecorated first, then target
+ else: #NOTE: reproduces the code!
+ org = getsource(func, enclosing=True, lstrip=False)
+ src = importable(fobj, alias, source=True) # include dependencies
+ src = (org, src) # target first, then decorated
+ func_vars[name] = src
+ src = ''.join(free_vars.values())
+ if not func_vars: #FIXME: 'enclose' in dummy; wrong ref 'name'
+ org = getsource(func, alias, force=True, enclosing=False, lstrip=True)
+ src = (src, org) # variables first, then target
+ else:
+ src = (src, None) # just variables (better '' instead of None?)
+ func_vars[None] = src
+ # FIXME: remove duplicates (however, order is important...)
+ return func_vars
+
+def importable(obj, alias='', source=None, builtin=True):
+ """get an importable string (i.e. source code or the import string)
+ for the given object, including any required objects from the enclosing
+ and global scope
+
+ This function will attempt to discover the name of the object, or the repr
+ of the object, or the source code for the object. To attempt to force
+ discovery of the source code, use source=True, to attempt to force the
+ use of an import, use source=False; otherwise an import will be sought
+ for objects not defined in __main__. The intent is to build a string
+ that can be imported from a python file.
+
+ obj is the object to inspect. If alias is provided, then rename the
+ object with the given alias. If builtin=True, then force an import for
+ builtins where possible.
+ """
+ #NOTE: we always 'force', and 'lstrip' as necessary
+ #NOTE: for 'enclosing', use importable(outermost(obj))
+ if source is None:
+ source = True if isfrommain(obj) else False
+ elif builtin and isbuiltin(obj):
+ source = False
+ tried_source = tried_import = False
+ while True:
+ if not source: # we want an import
+ try:
+ if _isinstance(obj): # for instances, punt to _importable
+ return _importable(obj, alias, source=False, builtin=builtin)
+ src = _closuredimport(obj, alias=alias, builtin=builtin)
+ if len(src) == 0:
+ raise NotImplementedError('not implemented')
+ if len(src) > 1:
+ raise NotImplementedError('not implemented')
+ return list(src.values())[0]
+ except:
+ if tried_source: raise
+ tried_import = True
+ # we want the source
+ try:
+ src = _closuredsource(obj, alias=alias)
+ if len(src) == 0:
+ raise NotImplementedError('not implemented')
+ # groan... an inline code stitcher
+ def _code_stitcher(block):
+ "stitch together the strings in tuple 'block'"
+ if block[0] and block[-1]: block = '\n'.join(block)
+ elif block[0]: block = block[0]
+ elif block[-1]: block = block[-1]
+ else: block = ''
+ return block
+ # get free_vars first
+ _src = _code_stitcher(src.pop(None))
+ _src = [_src] if _src else []
+ # get func_vars
+ for xxx in src.values():
+ xxx = _code_stitcher(xxx)
+ if xxx: _src.append(xxx)
+ # make a single source string
+ if not len(_src):
+ src = ''
+ elif len(_src) == 1:
+ src = _src[0]
+ else:
+ src = '\n'.join(_src)
+ # get source code of objects referred to by obj in global scope
+ from dill.detect import globalvars
+ obj = globalvars(obj) #XXX: don't worry about alias?
+ obj = list(getsource(_obj,name,force=True) for (name,_obj) in obj.items())
+ obj = '\n'.join(obj) if obj else ''
+ # combine all referred-to source (global then enclosing)
+ if not obj: return src
+ if not src: return obj
+ return obj + src
+ except:
+ if tried_import: raise
+ tried_source = True
+ source = not source
+ # should never get here
+ return
+
+
+# backward compatability
+def getimportable(obj, alias='', byname=True, explicit=False):
+ return importable(obj,alias,source=(not byname),builtin=explicit)
+ #return outdent(_importable(obj,alias,source=(not byname),builtin=explicit))
+def likely_import(obj, passive=False, explicit=False):
+ return getimport(obj, verify=(not passive), builtin=explicit)
+def _likely_import(first, last, passive=False, explicit=True):
+ return _getimport(first, last, verify=(not passive), builtin=explicit)
+_get_name = getname
+getblocks_from_history = getblocks
+
+
+
+# EOF
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/temp.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/temp.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/temp.py
new file mode 100644
index 0000000..827efda
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/dill/temp.py
@@ -0,0 +1,254 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+#!/usr/bin/env python
+#
+# Author: Mike McKerns (mmckerns @caltech and @uqfoundation)
+# Copyright (c) 2008-2014 California Institute of Technology.
+# License: 3-clause BSD. The full license text is available at:
+# - http://trac.mystic.cacr.caltech.edu/project/pathos/browser/dill/LICENSE
+"""
+Methods for serialized objects (or source code) stored in temporary files
+and file-like objects.
+"""
+#XXX: better instead to have functions write to any given file-like object ?
+#XXX: currently, all file-like objects are created by the function...
+
+from __future__ import absolute_import
+__all__ = ['dump_source', 'dump', 'dumpIO_source', 'dumpIO',\
+ 'load_source', 'load', 'loadIO_source', 'loadIO']
+
+from .dill import PY3
+
+def b(x): # deal with b'foo' versus 'foo'
+ import codecs
+ return codecs.latin_1_encode(x)[0]
+
+def load_source(file, **kwds):
+ """load an object that was stored with dill.temp.dump_source
+
+ file: filehandle
+ alias: string name of stored object
+ mode: mode to open the file, one of: {'r', 'rb'}
+
+ >>> f = lambda x: x**2
+ >>> pyfile = dill.temp.dump_source(f, alias='_f')
+ >>> _f = dill.temp.load_source(pyfile)
+ >>> _f(4)
+ 16
+ """
+ alias = kwds.pop('alias', None)
+ mode = kwds.pop('mode', 'r')
+ fname = getattr(file, 'name', file) # fname=file.name or fname=file (if str)
+ source = open(fname, mode=mode, **kwds).read()
+ if not alias:
+ tag = source.strip().splitlines()[-1].split()
+ if tag[0] != '#NAME:':
+ stub = source.splitlines()[0]
+ raise IOError("unknown name for code: %s" % stub)
+ alias = tag[-1]
+ local = {}
+ exec(source, local)
+ _ = eval("%s" % alias, local)
+ return _
+
+def dump_source(object, **kwds):
+ """write object source to a NamedTemporaryFile (instead of dill.dump)
+Loads with "import" or "dill.temp.load_source". Returns the filehandle.
+
+ >>> f = lambda x: x**2
+ >>> pyfile = dill.temp.dump_source(f, alias='_f')
+ >>> _f = dill.temp.load_source(pyfile)
+ >>> _f(4)
+ 16
+
+ >>> f = lambda x: x**2
+ >>> pyfile = dill.temp.dump_source(f, dir='.')
+ >>> modulename = os.path.basename(pyfile.name).split('.py')[0]
+ >>> exec('from %s import f as _f' % modulename)
+ >>> _f(4)
+ 16
+
+Optional kwds:
+ If 'alias' is specified, the object will be renamed to the given string.
+
+ If 'prefix' is specified, the file name will begin with that prefix,
+ otherwise a default prefix is used.
+
+ If 'dir' is specified, the file will be created in that directory,
+ otherwise a default directory is used.
+
+ If 'text' is specified and true, the file is opened in text
+ mode. Else (the default) the file is opened in binary mode. On
+ some operating systems, this makes no difference.
+
+NOTE: Keep the return value for as long as you want your file to exist !
+ """ #XXX: write a "load_source"?
+ from .source import importable, getname
+ import tempfile
+ kwds.pop('suffix', '') # this is *always* '.py'
+ alias = kwds.pop('alias', '') #XXX: include an alias so a name is known
+ name = str(alias) or getname(object)
+ name = "\n#NAME: %s\n" % name
+ #XXX: assumes kwds['dir'] is writable and on $PYTHONPATH
+ file = tempfile.NamedTemporaryFile(suffix='.py', **kwds)
+ file.write(b(''.join([importable(object, alias=alias),name])))
+ file.flush()
+ return file
+
+def load(file, **kwds):
+ """load an object that was stored with dill.temp.dump
+
+ file: filehandle
+ mode: mode to open the file, one of: {'r', 'rb'}
+
+ >>> dumpfile = dill.temp.dump([1, 2, 3, 4, 5])
+ >>> dill.temp.load(dumpfile)
+ [1, 2, 3, 4, 5]
+ """
+ import dill as pickle
+ mode = kwds.pop('mode', 'rb')
+ name = getattr(file, 'name', file) # name=file.name or name=file (if str)
+ return pickle.load(open(name, mode=mode, **kwds))
+
+def dump(object, **kwds):
+ """dill.dump of object to a NamedTemporaryFile.
+Loads with "dill.temp.load". Returns the filehandle.
+
+ >>> dumpfile = dill.temp.dump([1, 2, 3, 4, 5])
+ >>> dill.temp.load(dumpfile)
+ [1, 2, 3, 4, 5]
+
+Optional kwds:
+ If 'suffix' is specified, the file name will end with that suffix,
+ otherwise there will be no suffix.
+
+ If 'prefix' is specified, the file name will begin with that prefix,
+ otherwise a default prefix is used.
+
+ If 'dir' is specified, the file will be created in that directory,
+ otherwise a default directory is used.
+
+ If 'text' is specified and true, the file is opened in text
+ mode. Else (the default) the file is opened in binary mode. On
+ some operating systems, this makes no difference.
+
+NOTE: Keep the return value for as long as you want your file to exist !
+ """
+ import dill as pickle
+ import tempfile
+ file = tempfile.NamedTemporaryFile(**kwds)
+ pickle.dump(object, file)
+ file.flush()
+ return file
+
+def loadIO(buffer, **kwds):
+ """load an object that was stored with dill.temp.dumpIO
+
+ buffer: buffer object
+
+ >>> dumpfile = dill.temp.dumpIO([1, 2, 3, 4, 5])
+ >>> dill.temp.loadIO(dumpfile)
+ [1, 2, 3, 4, 5]
+ """
+ import dill as pickle
+ if PY3:
+ from io import BytesIO as StringIO
+ else:
+ from StringIO import StringIO
+ value = getattr(buffer, 'getvalue', buffer) # value or buffer.getvalue
+ if value != buffer: value = value() # buffer.getvalue()
+ return pickle.load(StringIO(value))
+
+def dumpIO(object, **kwds):
+ """dill.dump of object to a buffer.
+Loads with "dill.temp.loadIO". Returns the buffer object.
+
+ >>> dumpfile = dill.temp.dumpIO([1, 2, 3, 4, 5])
+ >>> dill.temp.loadIO(dumpfile)
+ [1, 2, 3, 4, 5]
+ """
+ import dill as pickle
+ if PY3:
+ from io import BytesIO as StringIO
+ else:
+ from StringIO import StringIO
+ file = StringIO()
+ pickle.dump(object, file)
+ file.flush()
+ return file
+
+def loadIO_source(buffer, **kwds):
+ """load an object that was stored with dill.temp.dumpIO_source
+
+ buffer: buffer object
+ alias: string name of stored object
+
+ >>> f = lambda x:x**2
+ >>> pyfile = dill.temp.dumpIO_source(f, alias='_f')
+ >>> _f = dill.temp.loadIO_source(pyfile)
+ >>> _f(4)
+ 16
+ """
+ alias = kwds.pop('alias', None)
+ source = getattr(buffer, 'getvalue', buffer) # source or buffer.getvalue
+ if source != buffer: source = source() # buffer.getvalue()
+ if PY3: source = source.decode() # buffer to string
+ if not alias:
+ tag = source.strip().splitlines()[-1].split()
+ if tag[0] != '#NAME:':
+ stub = source.splitlines()[0]
+ raise IOError("unknown name for code: %s" % stub)
+ alias = tag[-1]
+ local = {}
+ exec(source, local)
+ _ = eval("%s" % alias, local)
+ return _
+
+def dumpIO_source(object, **kwds):
+ """write object source to a buffer (instead of dill.dump)
+Loads by with dill.temp.loadIO_source. Returns the buffer object.
+
+ >>> f = lambda x:x**2
+ >>> pyfile = dill.temp.dumpIO_source(f, alias='_f')
+ >>> _f = dill.temp.loadIO_source(pyfile)
+ >>> _f(4)
+ 16
+
+Optional kwds:
+ If 'alias' is specified, the object will be renamed to the given string.
+ """
+ from .source import importable, getname
+ if PY3:
+ from io import BytesIO as StringIO
+ else:
+ from StringIO import StringIO
+ alias = kwds.pop('alias', '') #XXX: include an alias so a name is known
+ name = str(alias) or getname(object)
+ name = "\n#NAME: %s\n" % name
+ #XXX: assumes kwds['dir'] is writable and on $PYTHONPATH
+ file = StringIO()
+ file.write(b(''.join([importable(object, alias=alias),name])))
+ file.flush()
+ return file
+
+
+del absolute_import
+
+
+# EOF
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/executor.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/executor.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/executor.py
new file mode 100644
index 0000000..9d80c82
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/executor.py
@@ -0,0 +1,58 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import sys
+import socket
+import struct
+#argv[1] = port
+
+
+try:
+ import dill
+ port = int(sys.argv[1])
+
+ s1 = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
+ s1.bind((socket.gethostbyname("localhost"), 0))
+ s1.sendto(struct.pack(">i", s1.getsockname()[1]), (socket.gethostbyname("localhost"), port))
+
+ size = struct.unpack(">i", s1.recv(4))[0]
+ serialized_operator = s1.recv(size)
+
+ size = struct.unpack(">i", s1.recv(4))[0]
+ import_string = s1.recv(size).decode("utf-8")
+
+ size = struct.unpack(">i", s1.recv(4))[0]
+ input_file = s1.recv(size).decode("utf-8")
+
+ size = struct.unpack(">i", s1.recv(4))[0]
+ output_file = s1.recv(size).decode("utf-8")
+
+ exec(import_string)
+
+ operator = dill.loads(serialized_operator)
+ operator._configure(input_file, output_file, port)
+ operator._go()
+ sys.stdout.flush()
+ sys.stderr.flush()
+except:
+ sys.stdout.flush()
+ sys.stderr.flush()
+ s = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
+ s.bind((socket.gethostbyname("localhost"), 0))
+ destination = (socket.gethostbyname("localhost"), int(sys.argv[1]))
+ s.sendto(struct.pack(">i", -2), destination)
+ raise
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/__init__.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/__init__.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/__init__.py
new file mode 100644
index 0000000..d35bf39
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/__init__.py
@@ -0,0 +1,17 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Collector.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Collector.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Collector.py
new file mode 100644
index 0000000..88ada88
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Collector.py
@@ -0,0 +1,161 @@
+# ###############################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+from struct import pack
+import sys
+
+from flink.connection.Constants import Types
+
+PY2 = sys.version_info[0] == 2
+PY3 = sys.version_info[0] == 3
+
+if PY2:
+ stringtype = basestring
+else:
+ stringtype = str
+
+
+class Collector(object):
+ def __init__(self, con):
+ self._connection = con
+ self._serializer = None
+
+ def _close(self):
+ self._connection.send_end_signal()
+
+ def collect(self, value):
+ self._serializer = _get_serializer(self._connection.write, value)
+ self.collect = self._collect
+ self.collect(value)
+
+ def _collect(self, value):
+ self._connection.write(self._serializer.serialize(value))
+
+
+def _get_serializer(write, value):
+ if isinstance(value, (list, tuple)):
+ write(Types.TYPE_TUPLE)
+ write(pack(">I", len(value)))
+ return TupleSerializer(write, value)
+ elif value is None:
+ write(Types.TYPE_NULL)
+ return NullSerializer()
+ elif isinstance(value, stringtype):
+ write(Types.TYPE_STRING)
+ return StringSerializer()
+ elif isinstance(value, bool):
+ write(Types.TYPE_BOOLEAN)
+ return BooleanSerializer()
+ elif isinstance(value, int) or PY2 and isinstance(value, long):
+ write(Types.TYPE_LONG)
+ return LongSerializer()
+ elif isinstance(value, bytearray):
+ write(Types.TYPE_BYTES)
+ return ByteArraySerializer()
+ elif isinstance(value, float):
+ write(Types.TYPE_DOUBLE)
+ return FloatSerializer()
+ else:
+ raise Exception("Unsupported Type encountered.")
+
+
+class TupleSerializer(object):
+ def __init__(self, write, value):
+ self.serializer = [_get_serializer(write, field) for field in value]
+
+ def serialize(self, value):
+ bits = []
+ for i in range(len(value)):
+ bits.append(self.serializer[i].serialize(value[i]))
+ return b"".join(bits)
+
+
+class BooleanSerializer(object):
+ def serialize(self, value):
+ return pack(">?", value)
+
+
+class FloatSerializer(object):
+ def serialize(self, value):
+ return pack(">d", value)
+
+
+class LongSerializer(object):
+ def serialize(self, value):
+ return pack(">q", value)
+
+
+class ByteArraySerializer(object):
+ def serialize(self, value):
+ value = bytes(value)
+ return pack(">I", len(value)) + value
+
+
+class StringSerializer(object):
+ def serialize(self, value):
+ value = value.encode("utf-8")
+ return pack(">I", len(value)) + value
+
+
+class NullSerializer(object):
+ def serialize(self, value):
+ return b""
+
+
+class TypedCollector(object):
+ def __init__(self, con):
+ self._connection = con
+
+ def collectBytes(self, value):
+ size = pack(">I", len(value))
+ self._connection.write(b"".join([Types.TYPE_BYTES, size, value]))
+
+ def collect(self, value):
+ if not isinstance(value, (list, tuple)):
+ self._send_field(value)
+ else:
+ self._connection.write(Types.TYPE_TUPLE)
+ meta = pack(">I", len(value))
+ self._connection.write(bytes([meta[3]]) if PY3 else meta[3])
+ for field in value:
+ self.collect(field)
+
+ def _send_field(self, value):
+ if value is None:
+ self._connection.write(Types.TYPE_NULL)
+ elif isinstance(value, stringtype):
+ value = value.encode("utf-8")
+ size = pack(">I", len(value))
+ self._connection.write(b"".join([Types.TYPE_STRING, size, value]))
+ elif isinstance(value, bytes):
+ size = pack(">I", len(value))
+ self._connection.write(b"".join([Types.TYPE_BYTES, size, value]))
+ elif isinstance(value, bool):
+ data = pack(">?", value)
+ self._connection.write(b"".join([Types.TYPE_BOOLEAN, data]))
+ elif isinstance(value, int) or PY2 and isinstance(value, long):
+ data = pack(">q", value)
+ self._connection.write(b"".join([Types.TYPE_LONG, data]))
+ elif isinstance(value, float):
+ data = pack(">d", value)
+ self._connection.write(b"".join([Types.TYPE_DOUBLE, data]))
+ elif isinstance(value, bytearray):
+ value = bytes(value)
+ size = pack(">I", len(value))
+ self._connection.write(b"".join([Types.TYPE_BYTES, size, value]))
+ else:
+ raise Exception("Unsupported Type encountered.")
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Connection.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Connection.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Connection.py
new file mode 100644
index 0000000..7d64352
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Connection.py
@@ -0,0 +1,172 @@
+# ###############################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import mmap
+import socket
+import tempfile
+from struct import pack, unpack
+from collections import deque
+import sys
+PY2 = sys.version_info[0] == 2
+PY3 = sys.version_info[0] == 3
+
+MAPPED_FILE_SIZE = 1024 * 1024 * 64
+
+SIGNAL_REQUEST_BUFFER = b"\x00\x00\x00\x00"
+SIGNAL_REQUEST_BUFFER_G0 = b"\xFF\xFF\xFF\xFD"
+SIGNAL_REQUEST_BUFFER_G1 = b"\xFF\xFF\xFF\xFC"
+SIGNAL_FINISHED = b"\xFF\xFF\xFF\xFF"
+
+if PY2:
+ SIGNAL_WAS_LAST = "\x20"
+else:
+ SIGNAL_WAS_LAST = 32
+
+
+class OneWayBusyBufferingMappedFileConnection(object):
+ def __init__(self):
+ self._output_file = open(tempfile.gettempdir() + "/flink_data/output", "rb+")
+ self._file_output_buffer = mmap.mmap(self._output_file.fileno(), MAPPED_FILE_SIZE, mmap.MAP_SHARED, mmap.ACCESS_WRITE)
+
+ self._out = deque()
+ self._out_size = 0
+
+ self._offset_limit = MAPPED_FILE_SIZE - 1024 * 1024 * 3
+
+ def write(self, msg):
+ self._out.append(msg)
+ self._out_size += len(msg)
+ if self._out_size > self._offset_limit:
+ self._write_buffer()
+
+ def _write_buffer(self):
+ self._file_output_buffer.seek(1, 0)
+ self._file_output_buffer.write(b"".join(self._out))
+ self._file_output_buffer.seek(0, 0)
+ self._file_output_buffer.write(b'\x01')
+
+
+class BufferingUDPMappedFileConnection(object):
+ def __init__(self, input_file=tempfile.gettempdir() + "/flink_data/input", output_file=tempfile.gettempdir() + "/flink_data/output", port=25000):
+ self._input_file = open(input_file, "rb+")
+ self._output_file = open(output_file, "rb+")
+ self._file_input_buffer = mmap.mmap(self._input_file.fileno(), MAPPED_FILE_SIZE, mmap.MAP_SHARED, mmap.ACCESS_READ)
+ self._file_output_buffer = mmap.mmap(self._output_file.fileno(), MAPPED_FILE_SIZE, mmap.MAP_SHARED, mmap.ACCESS_WRITE)
+ self._socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
+ self._socket.bind((socket.gethostbyname("localhost"), 0))
+ self._socket.settimeout(300)
+ self._destination = (socket.gethostbyname("localhost"), port)
+
+ self._socket.sendto(pack(">I", self._socket.getsockname()[1]), self._destination)
+ self._socket.sendto(pack(">I", self._socket.getsockname()[1]), self._destination)
+
+ self._out = deque()
+ self._out_size = 0
+
+ self._input = ""
+ self._input_offset = 0
+ self._input_size = 0
+ self._was_last = False
+
+ def write(self, msg):
+ length = len(msg)
+ if length > MAPPED_FILE_SIZE:
+ raise Exception("Serialized object does not fit into a single buffer.")
+ tmp = self._out_size + length
+ if tmp > MAPPED_FILE_SIZE:
+ self._write_buffer()
+ self.write(msg)
+ else:
+ self._out.append(msg)
+ self._out_size = tmp
+
+ def _write_buffer(self):
+ self._file_output_buffer.seek(0, 0)
+ self._file_output_buffer.write(b"".join(self._out))
+ self._socket.sendto(pack(">i", self._out_size), self._destination)
+ self._out.clear()
+ self._out_size = 0
+ self._socket.recvfrom(1)
+
+ def read(self, des_size, ignored=None):
+ if self._input_size == self._input_offset:
+ self._read_buffer()
+ old_offset = self._input_offset
+ self._input_offset += des_size
+ return self._input[old_offset:self._input_offset]
+
+ def _read_buffer(self):
+ self._socket.sendto(SIGNAL_REQUEST_BUFFER, self._destination)
+ self._file_input_buffer.seek(0, 0)
+ self._input_offset = 0
+ meta_size = self._socket.recvfrom(5)[0]
+ self._input_size = unpack(">I", meta_size[:4])[0]
+ self._was_last = meta_size[4] == SIGNAL_WAS_LAST
+ self._input = self._file_input_buffer.read(self._input_size)
+
+ def send_end_signal(self):
+ if self._out_size:
+ self._write_buffer()
+ self._socket.sendto(SIGNAL_FINISHED, self._destination)
+
+ def has_next(self, ignored=None):
+ return not self._was_last or not self._input_size == self._input_offset
+
+ def reset(self):
+ self._was_last = False
+ self._input_size = 0
+ self._input_offset = 0
+ self._input = ""
+
+
+class TwinBufferingUDPMappedFileConnection(BufferingUDPMappedFileConnection):
+ def __init__(self, input_file=tempfile.gettempdir() + "/flink/data/input", output_file=tempfile.gettempdir() + "/flink/data/output", port=25000):
+ super(TwinBufferingUDPMappedFileConnection, self).__init__(input_file, output_file, port)
+ self._input = ["", ""]
+ self._input_offset = [0, 0]
+ self._input_size = [0, 0]
+ self._was_last = [False, False]
+
+ def read(self, des_size, group):
+ if self._input_size[group] == self._input_offset[group]:
+ self._read_buffer(group)
+ old_offset = self._input_offset[group]
+ self._input_offset[group] += des_size
+ return self._input[group][old_offset:self._input_offset[group]]
+
+ def _read_buffer(self, group):
+ if group:
+ self._socket.sendto(SIGNAL_REQUEST_BUFFER_G1, self._destination)
+ else:
+ self._socket.sendto(SIGNAL_REQUEST_BUFFER_G0, self._destination)
+ self._file_input_buffer.seek(0, 0)
+ self._input_offset[group] = 0
+ meta_size = self._socket.recvfrom(5)[0]
+ self._input_size[group] = unpack(">I", meta_size[:4])[0]
+ self._was_last[group] = meta_size[4] == SIGNAL_WAS_LAST
+ self._input[group] = self._file_input_buffer.read(self._input_size[group])
+
+ def has_next(self, group):
+ return not self._was_last[group] or not self._input_size[group] == self._input_offset[group]
+
+ def reset(self):
+ self._was_last = [False, False]
+ self._input_size = [0, 0]
+ self._input_offset = [0, 0]
+ self._input = ["", ""]
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/d182daa1/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Constants.py
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Constants.py b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Constants.py
new file mode 100644
index 0000000..0ca2232
--- /dev/null
+++ b/flink-staging/flink-language-binding/flink-python/src/main/python/org/apache/flink/languagebinding/api/python/flink/connection/Constants.py
@@ -0,0 +1,31 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+
+class Types(object):
+ TYPE_TUPLE = b'\x0B'
+ TYPE_BOOLEAN = b'\x0A'
+ TYPE_BYTE = b'\x09'
+ TYPE_SHORT = b'\x08'
+ TYPE_INTEGER = b'\x07'
+ TYPE_LONG = b'\x06'
+ TYPE_DOUBLE = b'\x04'
+ TYPE_FLOAT = b'\x05'
+ TYPE_STRING = b'\x02'
+ TYPE_NULL = b'\x00'
+ TYPE_BYTES = b'\x01'