<s> """<STR_LIT>""" <EOL> import contextlib <EOL> from logging import debug <EOL> from nuitka import Tracing , VariableRegistry <EOL> from nuitka . __past__ import iterItems <EOL> from nuitka . containers . oset import OrderedSet <EOL> from nuitka . importing . ImportCache import ( <EOL> getImportedModuleByName , <EOL> isImportedModuleByName <EOL> ) <EOL> from nuitka . ModuleRegistry import addUsedModule <EOL> from nuitka . nodes . NodeMakingHelpers import getComputationResult <EOL> from nuitka . PythonVersions import python_version <EOL> from . VariableTraces import ( <EOL> VariableTraceAssign , <EOL> VariableTraceInit , <EOL> VariableTraceLoopMerge , <EOL> VariableTraceMerge , <EOL> VariableTraceUninit , <EOL> VariableTraceUnknown <EOL> ) <EOL> signalChange = None <EOL> class VariableUsageTrackingMixin : <EOL> def initVariable ( self , variable ) : <EOL> if variable . isParameterVariable ( ) : <EOL> result = self . _initVariableInit ( variable ) <EOL> elif variable . isLocalVariable ( ) : <EOL> result = self . _initVariableUninit ( variable ) <EOL> elif variable . isMaybeLocalVariable ( ) : <EOL> result = self . _initVariableUnknown ( variable ) <EOL> elif variable . isModuleVariable ( ) : <EOL> result = self . _initVariableUnknown ( variable ) <EOL> elif variable . isTempVariable ( ) : <EOL> result = self . _initVariableUninit ( variable ) <EOL> else : <EOL> assert False , variable <EOL> assert result . getVariable ( ) is variable <EOL> return result <EOL> class CollectionTracingMixin : <EOL> def __init__ ( self ) : <EOL> self . variable_actives = { } <EOL> def getVariableCurrentTrace ( self , variable ) : <EOL> return self . getVariableTrace ( <EOL> variable = variable , <EOL> version = self . getCurrentVariableVersion ( variable ) <EOL> ) <EOL> def markCurrentVariableTrace ( self , variable , version ) : <EOL> self . variable_actives [ variable ] = version <EOL> def getCurrentVariableVersion ( self , variable ) : <EOL> try : <EOL> return self . variable_actives [ variable ] <EOL> except KeyError : <EOL> if not self . hasVariableTrace ( variable , <NUM_LIT:0> ) : <EOL> self . initVariable ( variable ) <EOL> self . markCurrentVariableTrace ( variable , <NUM_LIT:0> ) <EOL> return self . variable_actives [ variable ] <EOL> def getActiveVariables ( self ) : <EOL> return self . variable_actives . keys ( ) <EOL> def markActiveVariableAsUnknown ( self , variable ) : <EOL> current = self . getVariableCurrentTrace ( <EOL> variable = variable , <EOL> ) <EOL> if not current . isUnknownTrace ( ) : <EOL> version = variable . allocateTargetNumber ( ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = version , <EOL> trace = VariableTraceUnknown ( <EOL> owner = self . owner , <EOL> variable = variable , <EOL> version = version , <EOL> previous = current <EOL> ) <EOL> ) <EOL> self . markCurrentVariableTrace ( variable , version ) <EOL> def markActiveVariableAsLoopMerge ( self , variable ) : <EOL> current = self . getVariableCurrentTrace ( <EOL> variable = variable , <EOL> ) <EOL> version = variable . allocateTargetNumber ( ) <EOL> result = VariableTraceLoopMerge ( <EOL> variable = variable , <EOL> version = version , <EOL> previous = current <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = version , <EOL> trace = result <EOL> ) <EOL> self . markCurrentVariableTrace ( variable , version ) <EOL> return result <EOL> def markActiveVariablesAsUnknown ( self ) : <EOL> for variable in self . getActiveVariables ( ) : <EOL> self . markActiveVariableAsUnknown ( variable ) <EOL> class CollectionStartpointMixin : <EOL> def __init__ ( self ) : <EOL> self . variable_versions = { } <EOL> self . variable_traces = { } <EOL> self . break_collections = None <EOL> self . continue_collections = None <EOL> self . return_collections = None <EOL> self . exception_collections = None <EOL> def getLoopBreakCollections ( self ) : <EOL> return self . break_collections <EOL> def onLoopBreak ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> self . break_collections . append ( <EOL> ConstraintCollectionBranch ( <EOL> parent = collection , <EOL> name = "<STR_LIT>" <EOL> ) <EOL> ) <EOL> def getLoopContinueCollections ( self ) : <EOL> return self . continue_collections <EOL> def onLoopContinue ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> self . continue_collections . append ( <EOL> ConstraintCollectionBranch ( <EOL> parent = collection , <EOL> name = "<STR_LIT>" <EOL> ) <EOL> ) <EOL> def onFunctionReturn ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> if self . return_collections is not None : <EOL> self . return_collections . append ( <EOL> ConstraintCollectionBranch ( <EOL> parent = collection , <EOL> name = "<STR_LIT>" <EOL> ) <EOL> ) <EOL> def onExceptionRaiseExit ( self , raisable_exceptions , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> if self . exception_collections is not None : <EOL> self . exception_collections . append ( <EOL> ConstraintCollectionBranch ( <EOL> parent = collection , <EOL> name = "<STR_LIT>" <EOL> ) <EOL> ) <EOL> def getFunctionReturnCollections ( self ) : <EOL> return self . return_collections <EOL> def getExceptionRaiseCollections ( self ) : <EOL> return self . exception_collections <EOL> def hasVariableTrace ( self , variable , version ) : <EOL> return ( variable , version ) in self . variable_traces <EOL> def getVariableTrace ( self , variable , version ) : <EOL> return self . variable_traces [ ( variable , version ) ] <EOL> def getVariableTraces ( self , variable ) : <EOL> result = [ ] <EOL> for key , variable_trace in iterItems ( self . variable_traces ) : <EOL> candidate = key [ <NUM_LIT:0> ] <EOL> if variable is candidate : <EOL> result . append ( variable_trace ) <EOL> return result <EOL> def getVariableTracesAll ( self ) : <EOL> return self . variable_traces <EOL> def addVariableTrace ( self , variable , version , trace ) : <EOL> key = variable , version <EOL> assert key not in self . variable_traces , ( key , self ) <EOL> self . variable_traces [ key ] = trace <EOL> def addVariableMergeMultipleTrace ( self , variable , traces ) : <EOL> version = variable . allocateTargetNumber ( ) <EOL> trace_merge = VariableTraceMerge ( <EOL> variable = variable , <EOL> version = version , <EOL> traces = traces <EOL> ) <EOL> self . addVariableTrace ( variable , version , trace_merge ) <EOL> return version <EOL> def dumpTraces ( self ) : <EOL> debug ( "<STR_LIT>" , self ) <EOL> for _variable_desc , variable_trace in sorted ( iterItems ( self . variable_traces ) ) : <EOL> variable_trace . dump ( ) <EOL> def dumpActiveTraces ( self ) : <EOL> Tracing . printSeparator ( ) <EOL> Tracing . printLine ( "<STR_LIT>" ) <EOL> for variable , _version in sorted ( self . variable_actives . iteritems ( ) ) : <EOL> self . getVariableCurrentTrace ( variable ) . dump ( ) <EOL> Tracing . printSeparator ( ) <EOL> def _initVariableUnknown ( self , variable ) : <EOL> trace = VariableTraceUnknown ( <EOL> owner = self . owner , <EOL> variable = variable , <EOL> version = <NUM_LIT:0> , <EOL> previous = None <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = <NUM_LIT:0> , <EOL> trace = trace <EOL> ) <EOL> return trace <EOL> def _initVariableInit ( self , variable ) : <EOL> trace = VariableTraceInit ( <EOL> owner = self . owner , <EOL> variable = variable , <EOL> version = <NUM_LIT:0> <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = <NUM_LIT:0> , <EOL> trace = trace <EOL> ) <EOL> return trace <EOL> def _initVariableUninit ( self , variable ) : <EOL> trace = VariableTraceUninit ( <EOL> owner = self . owner , <EOL> variable = variable , <EOL> version = <NUM_LIT:0> , <EOL> previous = None <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = <NUM_LIT:0> , <EOL> trace = trace <EOL> ) <EOL> return trace <EOL> def updateFromCollection ( self , old_collection ) : <EOL> VariableRegistry . updateFromCollection ( old_collection , self ) <EOL> @ contextlib . contextmanager <EOL> def makeAbortStackContext ( self , catch_breaks , catch_continues , <EOL> catch_returns , catch_exceptions ) : <EOL> if catch_breaks : <EOL> old_break_collections = self . break_collections <EOL> self . break_collections = [ ] <EOL> if catch_continues : <EOL> old_continue_collections = self . continue_collections <EOL> self . continue_collections = [ ] <EOL> if catch_returns : <EOL> old_return_collections = self . return_collections <EOL> self . return_collections = [ ] <EOL> if catch_exceptions : <EOL> old_exception_collections = self . exception_collections <EOL> self . exception_collections = [ ] <EOL> yield <EOL> if catch_breaks : <EOL> self . break_collections = old_break_collections <EOL> if catch_continues : <EOL> self . continue_collections = old_continue_collections <EOL> if catch_returns : <EOL> self . return_collections = old_return_collections <EOL> if catch_exceptions : <EOL> self . exception_collections = old_exception_collections <EOL> class ConstraintCollectionBase ( CollectionTracingMixin ) : <EOL> def __init__ ( self , owner , name , parent ) : <EOL> CollectionTracingMixin . __init__ ( self ) <EOL> self . owner = owner <EOL> self . parent = parent <EOL> self . name = name <EOL> self . removes_knowledge = False <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( <EOL> self . __class__ . __name__ , <EOL> self . name , <EOL> id ( self ) <EOL> ) <EOL> @ staticmethod <EOL> def signalChange ( tags , source_ref , message ) : <EOL> signalChange ( tags , source_ref , message ) <EOL> def onUsedModule ( self , module ) : <EOL> return self . parent . onUsedModule ( module ) <EOL> @ staticmethod <EOL> def mustAlias ( a , b ) : <EOL> if a . isExpressionVariableRef ( ) and b . isExpressionVariableRef ( ) : <EOL> return a . getVariable ( ) is b . getVariable ( ) <EOL> return False <EOL> @ staticmethod <EOL> def mustNotAlias ( a , b ) : <EOL> return False <EOL> def removeKnowledge ( self , node ) : <EOL> pass <EOL> def onControlFlowEscape ( self , node ) : <EOL> for variable in self . getActiveVariables ( ) : <EOL> if variable . isModuleVariable ( ) : <EOL> self . markActiveVariableAsUnknown ( variable ) <EOL> elif python_version >= <NUM_LIT> or variable . isSharedTechnically ( ) : <EOL> self . markActiveVariableAsUnknown ( variable ) <EOL> def removeAllKnowledge ( self ) : <EOL> self . removes_knowledge = True <EOL> self . markActiveVariablesAsUnknown ( ) <EOL> def getVariableTrace ( self , variable , version ) : <EOL> return self . parent . getVariableTrace ( variable , version ) <EOL> def hasVariableTrace ( self , variable , version ) : <EOL> return self . parent . hasVariableTrace ( variable , version ) <EOL> def addVariableTrace ( self , variable , version , trace ) : <EOL> self . parent . addVariableTrace ( variable , version , trace ) <EOL> def addVariableMergeMultipleTrace ( self , variable , traces ) : <EOL> return self . parent . addVariableMergeMultipleTrace ( variable , traces ) <EOL> def onVariableSet ( self , assign_node ) : <EOL> variable_ref = assign_node . getTargetVariableRef ( ) <EOL> version = variable_ref . getVariableVersion ( ) <EOL> variable = variable_ref . getVariable ( ) <EOL> variable_trace = VariableTraceAssign ( <EOL> owner = self . owner , <EOL> assign_node = assign_node , <EOL> variable = variable , <EOL> version = version , <EOL> previous = self . getVariableCurrentTrace ( <EOL> variable = variable <EOL> ) <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = version , <EOL> trace = variable_trace <EOL> ) <EOL> self . markCurrentVariableTrace ( variable , version ) <EOL> return variable_trace <EOL> def onVariableDel ( self , variable_ref ) : <EOL> variable = variable_ref . getVariable ( ) <EOL> version = variable_ref . getVariableVersion ( ) <EOL> old_trace = self . getVariableCurrentTrace ( variable ) <EOL> variable_trace = VariableTraceUninit ( <EOL> owner = self . owner , <EOL> variable = variable , <EOL> version = version , <EOL> previous = old_trace <EOL> ) <EOL> self . addVariableTrace ( <EOL> variable = variable , <EOL> version = version , <EOL> trace = variable_trace <EOL> ) <EOL> self . markCurrentVariableTrace ( variable , version ) <EOL> def onLocalsUsage ( self ) : <EOL> for variable in self . getActiveVariables ( ) : <EOL> if not variable . isTempVariable ( ) and not variable . isModuleVariable ( ) : <EOL> variable_trace = self . getVariableCurrentTrace ( <EOL> variable <EOL> ) <EOL> variable_trace . addNameUsage ( ) <EOL> def onVariableRelease ( self , variable ) : <EOL> current = self . getVariableCurrentTrace ( variable ) <EOL> current . addRelease ( ) <EOL> return current <EOL> def onVariableContentEscapes ( self , variable ) : <EOL> self . getVariableCurrentTrace ( variable ) . onValueEscape ( ) <EOL> def onExpression ( self , expression , allow_none = False ) : <EOL> if expression is None and allow_none : <EOL> return None <EOL> assert expression . isExpression ( ) , expression <EOL> assert expression . parent , expression <EOL> r = expression . computeExpressionRaw ( <EOL> constraint_collection = self <EOL> ) <EOL> assert type ( r ) is tuple , expression <EOL> new_node , change_tags , change_desc = r <EOL> if change_tags is not None : <EOL> self . signalChange ( <EOL> change_tags , <EOL> expression . getSourceReference ( ) , <EOL> change_desc <EOL> ) <EOL> if new_node is not expression : <EOL> expression . replaceWith ( new_node ) <EOL> if new_node . isExpressionVariableRef ( ) or new_node . isExpressionTempVariableRef ( ) : <EOL> assert new_node . variable_trace . hasDefiniteUsages ( ) <EOL> if new_node . isExpressionVariableRef ( ) or new_node . isExpressionTempVariableRef ( ) : <EOL> if new_node . getVariable ( ) . isMaybeLocalVariable ( ) : <EOL> variable_trace = self . getVariableCurrentTrace ( <EOL> variable = new_node . getVariable ( ) . getMaybeVariable ( ) <EOL> ) <EOL> variable_trace . addUsage ( ) <EOL> return new_node <EOL> def onStatement ( self , statement ) : <EOL> try : <EOL> assert statement . isStatement ( ) , statement <EOL> new_statement , change_tags , change_desc = statement . computeStatement ( self ) <EOL> if new_statement is not statement : <EOL> self . signalChange ( <EOL> change_tags , <EOL> statement . getSourceReference ( ) , <EOL> change_desc <EOL> ) <EOL> return new_statement <EOL> except Exception : <EOL> Tracing . printError ( <EOL> "<STR_LIT>" % <EOL> statement . getSourceReference ( ) . getAsString ( ) <EOL> ) <EOL> raise <EOL> def mergeBranches ( self , collection_yes , collection_no ) : <EOL> """<STR_LIT>""" <EOL> if collection_yes is None and collection_no is None : <EOL> pass <EOL> elif collection_yes is None or collection_no is None : <EOL> self . mergeMultipleBranches ( <EOL> collections = ( self , collection_yes or collection_no ) <EOL> ) <EOL> else : <EOL> self . mergeMultipleBranches ( <EOL> collections = ( collection_yes , collection_no ) <EOL> ) <EOL> def mergeMultipleBranches ( self , collections ) : <EOL> assert len ( collections ) > <NUM_LIT:0> <EOL> if len ( collections ) == <NUM_LIT:1> : <EOL> self . replaceBranch ( collections [ <NUM_LIT:0> ] ) <EOL> return <EOL> variable_versions = { } <EOL> for collection in collections : <EOL> for variable , version in iterItems ( collection . variable_actives ) : <EOL> if variable not in variable_versions : <EOL> variable_versions [ variable ] = set ( [ version ] ) <EOL> else : <EOL> variable_versions [ variable ] . add ( version ) <EOL> for collection in collections : <EOL> for variable , versions in iterItems ( variable_versions ) : <EOL> if variable not in collection . variable_actives : <EOL> versions . add ( <NUM_LIT:0> ) <EOL> self . variable_actives = { } <EOL> for variable , versions in iterItems ( variable_versions ) : <EOL> if len ( versions ) == <NUM_LIT:1> : <EOL> version , = versions <EOL> else : <EOL> version = self . addVariableMergeMultipleTrace ( <EOL> variable = variable , <EOL> traces = [ <EOL> self . getVariableTrace ( variable , version ) <EOL> for version in <EOL> versions <EOL> ] <EOL> ) <EOL> self . markCurrentVariableTrace ( variable , version ) <EOL> def replaceBranch ( self , collection_replace ) : <EOL> self . variable_actives . update ( collection_replace . variable_actives ) <EOL> collection_replace . variable_actives = None <EOL> def onLoopBreak ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> return self . parent . onLoopBreak ( collection ) <EOL> def onLoopContinue ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> return self . parent . onLoopContinue ( collection ) <EOL> def onFunctionReturn ( self , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> return self . parent . onFunctionReturn ( collection ) <EOL> def onExceptionRaiseExit ( self , raisable_exceptions , collection = None ) : <EOL> if collection is None : <EOL> collection = self <EOL> return self . parent . onExceptionRaiseExit ( raisable_exceptions , collection ) <EOL> def getLoopBreakCollections ( self ) : <EOL> return self . parent . getLoopBreakCollections ( ) <EOL> def getLoopContinueCollections ( self ) : <EOL> return self . parent . getLoopContinueCollections ( ) <EOL> def getFunctionReturnCollections ( self ) : <EOL> return self . parent . getFunctionReturnCollections ( ) <EOL> def getExceptionRaiseCollections ( self ) : <EOL> return self . parent . getExceptionRaiseCollections ( ) <EOL> def makeAbortStackContext ( self , catch_breaks , catch_continues , <EOL> catch_returns , catch_exceptions ) : <EOL> return self . parent . makeAbortStackContext ( <EOL> catch_breaks = catch_breaks , <EOL> catch_continues = catch_continues , <EOL> catch_returns = catch_returns , <EOL> catch_exceptions = catch_exceptions <EOL> ) <EOL> def getCompileTimeComputationResult ( self , node , computation , description ) : <EOL> new_node , change_tags , message = getComputationResult ( <EOL> node = node , <EOL> computation = computation , <EOL> description = description <EOL> ) <EOL> if change_tags == "<STR_LIT>" : <EOL> self . onExceptionRaiseExit ( BaseException ) <EOL> return new_node , change_tags , message <EOL> class ConstraintCollectionBranch ( ConstraintCollectionBase ) : <EOL> def __init__ ( self , name , parent ) : <EOL> ConstraintCollectionBase . __init__ ( <EOL> self , <EOL> owner = parent . owner , <EOL> name = name , <EOL> parent = parent <EOL> ) <EOL> self . variable_actives = dict ( parent . variable_actives ) <EOL> def computeBranch ( self , branch ) : <EOL> if branch . isStatementsSequence ( ) : <EOL> result = branch . computeStatementsSequence ( <EOL> constraint_collection = self <EOL> ) <EOL> if result is not branch : <EOL> branch . replaceWith ( result ) <EOL> else : <EOL> self . onExpression ( <EOL> expression = branch <EOL> ) <EOL> def initVariable ( self , variable ) : <EOL> variable_trace = self . parent . initVariable ( variable ) <EOL> assert variable_trace . getVersion ( ) == <NUM_LIT:0> <EOL> self . variable_actives [ variable ] = <NUM_LIT:0> <EOL> return variable_trace <EOL> def dumpTraces ( self ) : <EOL> Tracing . printSeparator ( ) <EOL> self . parent . dumpTraces ( ) <EOL> Tracing . printSeparator ( ) <EOL> def dumpActiveTraces ( self ) : <EOL> Tracing . printSeparator ( ) <EOL> Tracing . printLine ( "<STR_LIT>" ) <EOL> for variable , _version in sorted ( self . variable_actives . iteritems ( ) ) : <EOL> self . getVariableCurrentTrace ( variable ) . dump ( ) <EOL> Tracing . printSeparator ( ) <EOL> class ConstraintCollectionFunction ( CollectionStartpointMixin , <EOL> ConstraintCollectionBase , <EOL> VariableUsageTrackingMixin ) : <EOL> def __init__ ( self , parent , function_body ) : <EOL> assert function_body . isExpressionFunctionBody ( ) or function_body . isExpressionClassBody ( ) or function_body . isExpressionGeneratorObjectBody ( ) or function_body . isExpressionCoroutineObjectBody ( ) , function_body <EOL> CollectionStartpointMixin . __init__ ( self ) <EOL> ConstraintCollectionBase . __init__ ( <EOL> self , <EOL> owner = function_body , <EOL> name = "<STR_LIT>" + str ( function_body ) , <EOL> parent = parent <EOL> ) <EOL> class ConstraintCollectionModule ( CollectionStartpointMixin , <EOL> ConstraintCollectionBase , <EOL> VariableUsageTrackingMixin ) : <EOL> def __init__ ( self , module ) : <EOL> CollectionStartpointMixin . __init__ ( self ) <EOL> ConstraintCollectionBase . __init__ ( <EOL> self , <EOL> owner = module , <EOL> name = "<STR_LIT>" , <EOL> parent = None <EOL> ) <EOL> self . used_modules = OrderedSet ( ) <EOL> def onUsedModule ( self , module_name ) : <EOL> self . used_modules . add ( module_name ) <EOL> if isImportedModuleByName ( module_name ) : <EOL> module = getImportedModuleByName ( module_name ) <EOL> addUsedModule ( module ) <EOL> def getUsedModules ( self ) : <EOL> return self . used_modules </s>
