Error in ' + h + ':
' + str(e) + '' )
editor.content.setAcceptRichText( False )
editor.content.setPlainText( value )
editor.connect( editor.content, SIGNAL( 'cursorPositionChanged()' ), f )
editor.btnOk.setText( 'Check and save as XHTML' )
editor.btnCancel.setText( 'Save as simple text' )
line = getattr( e, 'getLineNumber', None )
if line is not None:
line = line() - 2
else:
line = 0
column = getattr( e, 'getColumnNumber', None )
if column is not None:
column = column()
else:
column = 0
if line == 0 and column == 0:
x = re.match( '^[^0-9]*:([0-9]+):([0-9]+):', str(e) )
print x
print str(e)
if x:
line = int( x.group(1) ) - 3
column = int( x.group(2) ) # it's not the column !
# (maybe the XML tag number ?)
editor.content.moveCursor( QtGui.QTextCursor.Start )
for i in xrange( line - 1 ):
editor.content.moveCursor( QtGui.QTextCursor.NextBlock )
#for i in xrange( column - 1 ):
#editor.content.moveCursor( QtGui.QTextCursor.Right )
if editor.exec_() == QDialog.Accepted:
value = unicode( editor.content.toPlainText() )
goOn = True
else:
newValue = unicode( editor.content.toPlainText() )
goOn = False
d[ k ] = newValue
elif type( value ) is types.DictType:
stack += [ ( value, key, h + '.' + key ) for key in value.iterkeys() ]
#----------------------------------------------------------------------------
def getHTMLFileName( processId, documentation=None, language=None ):
"""
Gets the path to the html page corresponding to the documentation of the process in parameter.
"""
processInfo = getProcessInfo( processId )
if documentation is None:
documentation = readProcdoc( processId )
if language is None:
language = neuroConfig.language
htmlPath=XHTML.html(documentation.get( 'htmlPath', ''))
if htmlPath:
defaultPath=htmlPath
else:
defaultPath = os.path.dirname( neuroConfig.docPath )
return os.path.join( defaultPath, language, 'processes',
string.replace( processInfo.id, ' ', '_' ) + '.html' )
#----------------------------------------------------------------------------
def convertSpecialLinks( msg, language, baseForLinks, translator ):
"""
Converts special links and tags in a procdoc documentation.
The possible special links or tags are:
* *bvcategory://* refers to the documentation of a processes category (directory containing processes).
* *bvprocess://* refers to the documentation of a process
* *bvimage://* refers to an image in Brainvisa images directory
* *<_t_>* translates the string in the selected language
* * replaces the id by the name of the process
"""
stack = [ msg ]
while stack:
item = stack.pop()
if isinstance( item, XHTML ):# and \
stack += item.content
tag = item.tag
if not tag: continue
tag = tag.lower()
if tag == 'a':
href = item.attributes.get( 'href' )
if href is not None:
i = href.find( '#' )
if i >= 0:
postHref = href[ i: ]
href = href[ :i ]
else:
postHref = ''
if not href: continue
if href.startswith( 'bvcategory://' ):
href = href[ 13: ]
if href.startswith( '/' ):
href = href[ 1: ]
if baseForLinks:
base = baseForLinks + '/categories/'
else:
base = 'categories/'
href = base + href.lower() + '/category_documentation.html'
item.attributes[ 'href' ] = href + postHref
elif href.startswith( 'bvprocess://' ):
href = href[ 12: ]
if href.startswith( '/' ):
href = href[ 1: ]
if baseForLinks:
href = baseForLinks + '/' + href
href += '.html'
item.attributes[ 'href' ] = href + postHref
elif href.startswith( 'bvimage://' ):
href = href[ 10: ]
if href.startswith( '/' ):
href = href[ 1: ]
if baseForLinks:
href = baseForLinks + '/../../images/' + href
else:
href = '../../images/' + href
item.attributes[ 'href' ] = href
elif tag == 'img':
src = item.attributes.get( 'src', '' )
if not src: continue
elif src.startswith( 'bvimage://' ):
src = src[ 10: ]
if src.startswith( '/' ):
src = src[ 1: ]
if baseForLinks:
src = baseForLinks + '/../../images/' + src
else:
src = '../../images/' + src
item.attributes[ 'src' ] = src
elif tag == '_t_':
item.tag = None
item.content = ( translator.translate( item.content[0] ), )
elif tag == 'bvprocessname':
item.tag = None
try:
n = getProcessInfo( item.attributes[ 'name' ] ).name
except:
n = item.attributes[ 'name' ]
item.content = ( translator.translate( n, ) )
return msg
#----------------------------------------------------------------------------
def generateHTMLProcessesDocumentation( procId = None ):
"""
Generates HTML pages for the documentation of the process in parameter or for all processes if `procId` is None.
The process generateDocumentation is used.
"""
if procId is None:
defaultContext().runProcess("generateDocumentation")
else:
docproc = getProcessInstance( 'generateDocumentation' )
translators = {}
for l in neuroConfig._languages:
translators[ l ] = neuroConfig.Translator( l )
ontology = 'all'
docproc.generateHTMLDocumentation( procId, translators, defaultContext(),
ontology )
#----------------------------------------------------------------------------
def mapValuesToChildrenParameters(destNode, sourceNode, dest, source, value = None, defaultProcess = None, defaultProcessOptions = {}, name=None):
sourceObject, sourceParameter = sourceNode.parseParameterString( source )
l = getattr(sourceObject, sourceParameter, [])
lsize = len(l)
csize = len(destNode.childrenNames())
initcsize = csize
# Resulting size is the max between children size and list size
rsize = max(csize, lsize)
for i in xrange(rsize):
if i == csize:
if defaultProcess :
# Add a newc child node
child = brainvisa.processes.ProcessExecutionNode(
defaultProcess,
optional = defaultProcessOptions.get('optional', True),
selected = defaultProcessOptions.get('selected', True),
expandedInGui = defaultProcessOptions.get('expandedInGui', False)
)
destNode.addChild(name=name, node = child)
csize += 1
if i < lsize :
v = l[i]
else :
v = None
# Set node value
k = destNode.childrenNames()[i]
destChild = destNode._children[k]
destObject, destParameter = destChild.parseParameterString( dest )
setattr(destObject, destParameter, v)
i += 1
#----------------------------------------------------------------------------
def mapChildrenParametersToValues(destNode, sourceNode, dest, source, value = None):
r = []
for k in sourceNode.childrenNames():
sourceChild = sourceNode._children[k]
sourceObject, sourceParameter = sourceChild.parseParameterString( source )
s = getattr(sourceObject, sourceParameter, None)
r.append(s)
destObject, destParameter = destNode.parseParameterString( dest )
setattr(destObject, destParameter, r)
#----------------------------------------------------------------------------
class Parameterized( object ):
"""
This class represents an object that have a signature, that is to say a list of typed parameters.
A `Parameterized` object can notify the changes in its signature.
The parameters can be linked together, that is to say, if the value of a parameter changes, the value of all linked parameters may change.
This object has an :py:func:`initialization` function that can define the links between the parameters and their initial values.
:Attributes:
.. py:attribute:: signature
The signature is a :py:class:`brainvisa.data.neuroData.Signature`. It contains the list of parameters accepted by the object and their types. The possible types are described in :py:mod:`brainvisa.data.neuroData`.
.. py:attribute:: signatureChangeNotifier
This variable is a :py:class:`soma.notification.Notifier`. It calls its notify function when the signature of the :py:class:`Parameterized` object changes.
:Methods:
.. automethod:: initialization
.. automethod:: findValue
.. automethod:: setValue
.. automethod:: checkArguments
.. automethod:: setOptional
.. automethod:: isDefault
.. automethod:: setDefault
.. automethod:: setConvertedValue
.. automethod:: restoreConvertedValues
.. automethod:: parameterLinkable
.. automethod:: linkParameters
.. automethod:: addLink
.. automethod:: removeLink
.. automethod:: addParameterObserver
.. automethod:: removeParameterObserver
.. automethod:: changeSignature
.. automethod:: _parameterHasChanged
.. automethod:: clearLinksFrom
.. automethod:: clearLinksTo
.. automethod:: cleanup
"""
def __init__( self, signature ):
# print 'create Parameterized', self
self.__dict__[ '_deleted' ] = False # safety to avoid double deletion
# see http://code.activestate.com/lists/python-list/191512/
self.__dict__[ 'signature' ] = signature
self._convertedValues = {}
self._links = {}
self._isParameterSet = {}
self._isDefault = {}
self._immutable = {}
self._warn = {}
self.signatureChangeNotifier = Notifier( 1 )
self.deleteCallbacks = []
self._blocklinks = False
for i, p in self.signature.items():
np = copy.copy( p )
self.signature[ i ] = np
np.copyPostprocessing()
np.setNameAndParameterized( i, self )
# Values initialization
for i, p in self.signature.items():
self.setValue( i, p.defaultValue() )
self.initialization()
# Take into account links set during self.initialization() :
self.linksInitialization()
def linksInitialization( self, parameterizedObjects = None, params = None, excluded = None ):
if parameterizedObjects :
r = []
for o in parameterizedObjects :
if not isinstance(o, weakref.ProxyType):
o = weakref.proxy(o)
r.append(o)
parameterizedObjects = r
# Call parameterHasChanged for the parameters that have not their default value anymore or that have a not None value
for name in [n for n, v in self.signature.items() if ( (self.__dict__[n] != v.defaultValue()) or (self.__dict__[n] != None) ) ]:
if (not params or (name in params)) \
and (not excluded or (not name in excluded)) :
self._parameterHasChanged( name,
getattr( self, name ),
parameterizedObjects = parameterizedObjects )
def __del__( self ):
if not hasattr( self, '_deleted' ) or self._deleted:
return
# print 'del Parameterized', self
self._deleted = True
debugHere()
for x in self.deleteCallbacks:
x( self )
def _parameterHasChanged( self, name, newValue, parameterizedObjects = None ):
"""
This function is called when the value of an attribute described in the signature changes.
"""
debug = neuroConfig.debugParametersLinks
if debug: print >> debug, 'parameter', name, 'changed in', self, 'with value', newValue
for function in self._warn.get( name, [] ):
if debug: print >> debug, ' call (_warn)', function, '(', name, ',', newValue, ')'
function( self, name, newValue )
for parameterized, attribute, function, force in self._links.get( name, [] ):
if parameterized is None:
if debug: print >> debug, ' call (_links)', function, '(', self, ',', self, ')'
function( self, self )
elif (not parameterizedObjects) or (parameterized in parameterizedObjects):
if debug: print >> debug, ' ', name, 'is linked to parameter', attribute, 'of', parameterized, 'from', self, '(', len( self._links.get( name, [] ) ), ')'
linkParamType = parameterized.signature[ attribute ]
if not parameterized._isImmutable( attribute ) and ( force or parameterized.parameterLinkable( attribute, debug=debug ) ):
linkParamDebug = getattr( linkParamType, '_debug', None )
if linkParamDebug is not None:
print >> linkParamDebug, 'parameter', name, 'changed in', self, 'with value', newValue
if force:
parameterized.setDefault( attribute, self.isDefault( name ) )
if function is None:
if debug: print >> debug, ' ' + str(parameterized) + '.setValue(', repr(attribute), ',', newValue,')'
if linkParamDebug is not None:
print >> linkParamDebug, ' ==> ' + str(parameterized) + '.setValue(', repr(attribute), ',', newValue,')'
valueSet = newValue
parameterized.setValue( attribute, newValue )
else:
if debug: print >> debug, ' call', function, '(', parameterized, ',', self, ')'
if linkParamDebug is not None:
print >> linkParamDebug, ' ==> call', function, '(', parameterized, ',', self, ')'
v = function( parameterized, self )
valueSet=v
if debug: print >> debug, ' ' + str(parameterized) + '.setValue(', repr(attribute), ',', v,')'
if linkParamDebug is not None:
print >> linkParamDebug, ' ' + str(parameterized) + '.setValue(', repr(attribute), ',', v,')'
parameterized.setValue( attribute, v )
# activate the notifier with the parameter that receive a linked value and with the new value after evaluation of a link function.
parameterized.signature[ attribute ].valueLinkedNotifier.notify(
parameterized, attribute, valueSet )
if debug :
debug.flush()
def isDefault( self, key ):
"""Returns True if the parameter `key` has kept its default value."""
return self._isDefault.get( key, True )
def setDefault( self, key, value ):
"""Stores if the parameter `key` have kept its default value or not."""
debug = neuroConfig.debugParametersLinks
if debug: print >> debug, ' setDefault(', key, ',', value, ')'
self._isDefault[ key ] = value
def _isImmutable( self, key ):
"""Returns True if the parameter `key` is immutable, ie can really be changed by a link. This is an internal state, used only temporarily during parameters assignment."""
return self._immutable.get( key, False )
def _clearImmutableParameters( self ):
self._immutable = {}
def _setImmutable( self, key, value ):
"""Stores if the parameter `key` immutable, ie can really be changed by a link."""
self._immutable[ key ] = value
def parameterLinkable( self, key, debug=None ):
"""Indicates if the value of the parameter can change through a parameter link."""
if debug is None:
debug = neuroConfig.debugParametersLinks
result= bool( self.signature[ key ].linkParameterWithNonDefaultValue or \
self.isDefault( key ) )
if debug: print >> debug, ' parameterLinkable =', result
return result
def initialization( self ):
"""This function does nothing by default but it may be overrideed in processes classes to define initial values for the parameters or links between parameters."""
pass
def checkArguments( self ):
"""Checks the value of the parameters described in the signature."""
for p, o in self.signature.items():
o.checkValue( p, getattr( self, p, None ) )
def findValue( self, attributeName, value ):
"""Calls :py:func:`setValue`."""
self.setValue( attributeName, value )
def __setattr__( self, name, value ):
"""Calls :py:func:`setValue` if the parameter is described in the signature."""
if self.signature.has_key( name ):
self.setValue( name, value )
else:
self.__dict__[ name ] = value
def blockLinks( self, blocked ):
"""
While links are blocked, calls to setValue() or other parameters changes
do not trigger links.
"""
self._blocklinks = blocked
def setValue( self, name, value, default=None ):
"""
Checks the value, sets the attribute `name`.
If the value has changed, :py:meth:`_parameterHasChanged` is called to apply the links.
"""
debug = neuroConfig.debugParametersLinks
if debug:
print >> debug, str(self) + '.setValue(', repr(name), ',', repr(value), ',', repr(default), ')'
changed = False
if default is not None:
changed = self.isDefault( name ) != default
self.setDefault( name, default )
if self._isParameterSet.get( name, False ):
oldValue = getattr( self, name, None )
newValue = self.signature[ name ].findValue( value )
changed = changed or newValue != oldValue
else:
self._isParameterSet[ name ] = True
newValue = self.signature[ name ].findValue( value )
changed = True
self.__dict__[ name ] = newValue
if changed:
self._parameterHasChanged( name, newValue )
def linkParameters( self, destName, sources, function = None ):
"""
Links the parameters. When one of the `sources` parameters change, the value of `destName` parameter may change.
It is possible to give a specific link function that will be called when the link is applied but it is not mandatory, a default function exists according to the type of parameter.
:param string destName: name of the parameter that may change when the sources parameters change. If None, the link function will be called every time the sources parameters change.
:param sources: one or several parameters, whose modification will activate the link function.
:type sources: string, tuple or list
:param function function: specific function to call instead of the default one when the link is activated. The signature of the function is *function(self, process ) -> destination*
"""
if type( sources ) is types.StringType:
sourcesList = [ sources ]
else:
sourcesList = sources
for p in [ destName ] + list( sourcesList ):
if not self.signature.has_key( p ):
raise ValueError( HTMLMessage(_t_( '%s is not a valid parameter name' ) % p) )
if function is None:
function = getattr( self.signature[ destName ], 'defaultLinkParametersFunction', None )
for p in sourcesList:
self._links.setdefault( p, [] ).append( ( weakref.proxy( self ), destName, function, False ) )
def addParameterObserver( self, parameterName, function ):
"""Associates a callback function to the modifications of the parameter value.
:param parameterName: the name of the parameter whose modification will activate the callback.
:param function: the callback function. its signature is *function(self, parameterName, newValue)*
"""
minimum, maximum = numberOfParameterRange( function )
if maximum == 0:
tmp = lambda x, y, z, f=function: f()
tmp._save_function = function
function = tmp
self._warn.setdefault( parameterName, [] ).append( function )
def removeParameterObserver( self, parameterName, function ):
"""Removes the callback function from the parameter observers."""
l = self._warn.get( parameterName, None )
if l is not None:
l.remove( function )
if not l:
del self._warn[ parameterName ]
def setOptional( self, *args ):
"""Indicates that the parameters are not mandatory."""
for k in args:
self.signature[ k ].mandatory = False
def setMandatory( self, *args ):
"""Indicates that the parameters are mandatory."""
for k in args:
self.signature[ k ].mandatory = True
def setConvertedValue( self, name, value ):
"""Sets the value but stores the previous value in an internal dictionary."""
self._convertedValues[ name ] = getattr( self, name )
self.__dict__[ name ] = value
def restoreConvertedValues( self ):
"""Restore values as they were before conversions using the values stored in an internal dictionary."""
self.__dict__.update( self._convertedValues )
self._convertedValues.clear()
def addLink( self, destination, source, function=None ):
"""Add a link between `source` and `destination` parameters. When the value of `source` changes, the value of `destination` may change.
Contrary to :py:func:`linkParameters`, the link will always be applied, even if the `destination` parameter has no more its default value.
:param string destination: name of the parameter that may change when the source parameters change. If None, the link function will be called every time the source parameters change.
:param source: one or several parameters, whose modification will activate the link function.
:type source: string, tuple or list
:param function function: specific function that will be called instead of the default one when the link is activated. The signature of the function is *function(self, *sources ) -> destination*
"""
# Parse source
sources = []
if type( source ) in ( types.ListType, types.TupleType ):
for i in source:
if type( i ) in ( types.ListType, types.TupleType ):
sources.append( i )
else:
sources.append( ( self, i ) )
else:
sources.append( ( self, source ) )
if destination is None:
destObject, destParameter = ( None, None )
else:
destObject, destParameter = ( weakref.proxy( self ), destination )
# Check if a default function can be provided
if function is None:
if len( sources ) == 1:
function = lambda x: x
else:
raise RuntimeError( HTMLMessage(_t_( 'No function provided in addLink' )) )
multiLink = ExecutionNode.MultiParameterLink( sources, function )
for sourceObject, sourceParameter in sources:
sourceObject._links.setdefault( sourceParameter, [] ).append (
( destObject, destParameter, multiLink, True ) )
def removeLink( self, destination, source ):
"""Removes a link added with :py:func:`addLink` function."""
# print 'removeLink', self, destination, source
# Parse source
sources = []
if type( source ) in ( types.ListType, types.TupleType ):
for i in source:
sources.append( ( self, i ) )
else:
sources.append( ( self, source ) )
if destination is None:
destObject, destParameter = ( None, None )
else:
destObject, destParameter = ( self, destination )
removed = False
for sourceObject, sourceParameter in sources:
l = sourceObject._links.get( sourceParameter, [] )
if l:
lbis = l
l = [i for i in l if ( i[0] is not destObject and i[0] is not weakref.proxy( destObject ) ) or i[1] != destParameter]
if len( lbis ) != len( l ):
removed = True
if l:
sourceObject._links[ sourceParameter ] = l
else:
del sourceObject._links[ sourceParameter ]
else:
print 'warning: link not removed:', self, destination, 'from:', source
return removed
def changeSignature( self, signature ):
"""Sets a new signature. Previous values of attributes are kept if the attributes are still in the signature.
Links and observer callbacks that are no more associated to the signature parameters are deleted.
The :py:attr:`signatureChangeNotifier` is notified.
"""
# Change signature
self.signature = signature
for n in self.signature.keys():
setattr( self, n, getattr( self, n, None ) )
# Remove unused links
for n in self._links.keys():
if not self.signature.has_key( n ):
del self._links[ n ]
for n in self._warn.keys():
if not self.signature.has_key( n ):
del self._warn[ n ]
# Notify listeners
self.signatureChangeNotifier.notify( self )
def clearLinksTo( self, *args ):
"""Removes all links that have a parameter in `args` as a destination."""
for i in args:
if isinstance( i, basestring ):
destObject, destParameter = None, i
else:
destObject, destParameter = i
if destObject:
do=destObject
else:
do= self
#do = (self if destObject is None else destObject) # not work in python 2.4
if not do.signature.has_key( destParameter ):
raise KeyError( _t_( 'Object %(object)s has not parameter "%(param)s"' ) % { 'object': unicode( do ), 'param': destParameter } )
for k, l in self._links.items():
i = 0
while i < len( l ):
do, dp, ml, f = l[ i ]
if ( destObject is None or destObject is do ) and \
destParameter == dp:
del l[ i ]
else:
i += 1
def clearLinksFrom( self, *args ):
"""Removes all links associated to a parameter in `args` as a source. """
for k in args:
if self._links.has_key( k ):
del self._links[ k ]
def cleanup( self ):
"""Removes all links, observers, and stored converted values, reinitializes the signature change notifier."""
if debugHere is not None: # at exit time, the debug module might already be gone
debugHere()
self._convertedValues = {}
self._links = {}
self._warn = {}
self.signatureChangeNotifier = Notifier( 1 )
def convertStateValue( self, value ):
if value is not None and not isinstance( value, ( int, float, basestring, list, dict, tuple ) ):
result = unicode( value )
elif isinstance( value, list ):
result = [ self.convertStateValue( itervalue ) for itervalue in value ]
elif isinstance( value, tuple ):
result = tuple( self.convertStateValue( itervalue ) for itervalue in value )
elif isinstance( value, dict ) :
result = dict( (key, self.convertStateValue( itervalue ) ) for key, itervalue in value.iteritems() )
else :
result = value
return result
def saveStateInDictionary( self, result=None ):
if result is None:
result = {}
selected = {}
default = {}
for n in self.signature.iterkeys():
value = getattr( self, n, None )
value = self.convertStateValue( value )
if self.isDefault( n ):
default[ n ] = value
else:
selected[ n ] = value
result[ 'parameters' ] = {
'selected': selected,
'default': default,
}
return result
#----------------------------------------------------------------------------
class Process( Parameterized ):
"""
This class represents a Brainvisa process or pipeline.
This object has a **signature** that describes its inputs and outputs and an **execution function** :py:meth:`execution`.
If it is a **pipeline**, it also have an **execution node** that describes the structure of the pipeline.
:Attributes:
.. py:attribute:: signature
The signature is a :py:class:`brainvisa.data.neuroData.Signature`. It contains the list of parameters accepted by the object and their types. The possible types are described in :py:mod:`brainvisa.data.neuroData`.
.. py:attribute:: category (string)
The processes are organized into categories. Generally, the category is the name of the directory where the process file is located.
.. py:attribute:: userLevel (integer)
The process is available in Brainvisa interface if its userLevel is lower or equal than the userLevel selected in Brainvisa options.
0 : Basic, 1: Advanced, 2: Expert.
.. py:attribute:: showMaximized (boolean)
If true, the process window is shown maximized with a frame around it.
:Methods:
.. automethod:: id
.. automethod:: validation
.. automethod:: execution
.. automethod:: executionNode
.. automethod:: setExecutionNode
.. automethod:: getAllParameters
.. automethod:: allProcesses
.. automethod:: pipelineStructure
.. automethod:: sourceFile
.. automethod:: sourcePath
.. automethod:: inlineGUI
.. automethod:: _iterate
.. automethod:: _copy
"""
signature = Signature()
category = 'BrainVISA'
userLevel = 2
showMaximized = False
def __init__( self ):
# The following attributes can be set in user defined initialization()
# mathod which is called by Parameterized constructor. Therefore, it
# must be set before or never.
self._executionNode = None
# Copy signature because there is only one instance of each Parameter
# object in signature for each Process class. This is an old mistake in
# BrainVISA design, there should be one Signature instance by Process
# instance.
Parameterized.__init__( self, self.signature.shallowCopy())
self._log = None
self._outputLog = None
self._outputLogFile = None
#Main processes are opposed to subprocessed. There is more information
# displayed to the user on main processes. For example, start/stop
# notification and time elapsed in process are only displayed on main
# processes. By default all processes called by another process are not
# main process. It can be changed by setting isMainProcess to True.
self.isMainProcess = False
if hasattr( self.__class__, '_instance' ):
self.__class__._instance += 1
else:
self.__class__._instance = 1
self.instance = self.__class__._instance
def __del__( self ):
if self._deleted:
# print '*** Process already deleted ***'
return
try:
Parameterized.__del__( self )
except:
# can happen when quitting the application: the current module is
# not available any longer
pass
def _iterate( self, **kwargs ):
"""
Returns a list of copies of the current process with different parameters values.
:param kwargs: dictionary containing a list of values for each parameter name.
The first value is for the first process of the iteration and so on...
"""
# Find iteration size
requiredLength = 0
for values in kwargs.itervalues():
length = len( values )
if length > 1:
if requiredLength > 0 and length > 1 and requiredLength != length:
raise Exception( _t_( 'all lists of arguments with more than one value must have the same size' ) )
else:
requiredLength = length
# Set lists of values
finalValues = {}
for key, values in kwargs.iteritems():
if values:
if len( values ) == 1:
finalValues[ key ] = [ self.signature[ key ].findValue( values[0] ) ] * requiredLength
else:
finalValues[ key ] = [ self.signature[ key ].findValue( v ) for v in values ]
result = []
for i in xrange( requiredLength ):
p = self._copy( withparams=True ) # should copy only non-default params
for argumentName in finalValues.keys():
p._setImmutable( argumentName, True )
for argumentName, values in finalValues.iteritems():
p.setValue( argumentName, values[ i ], default=0 )
p._clearImmutableParameters()
result.append( p )
return result
def _copy( self, withparams=True ):
"""Returns a copy of the process. The value of the parameters are also copied if withparams is True (which is the default)"""
result = self.__class__()
if withparams:
# disable links
self.blockLinks( True )
# set params
for ( n, p ) in self.signature.items():
#if not self.isDefault( n ):
#result.setValue( n, getattr( self, n, None ), default=False )
result.setValue( n, getattr( self, n, None ),
default=self.isDefault( n ) )
if self._executionNode:
self._executionNode._copy(result.executionNode(), withparams=withparams)
self.blockLinks( False )
return result
def inlineGUI( self, values, context, parent, externalRunButton=False ):
"""This method can be overrideed in order to specialize buttons of the process window.
:param context: the execution context of the process
:param parent: The parent widget
:returns: the widget containing the buttons that will replace the default buttons (Run and Iterate)
:rtype: QWidget
"""
return None
def validation( self ):
"""This method can be overrideed in order to check if the process dependencies are available.
It will be called at Brainvisa startup when the processes are loaded. If the method raises an exception, the process will not be available.
:raises: :py:class:`brainvisa.validation.ValidationError`
"""
return 1
def id( self ):
"""Returns the process id."""
return self._id
def sourceFile( self ):
"""Returns the name of the source file of the process."""
return self._fileName
def sourcePath( self ):
"""Returns the path to the source file of the process."""
return os.path.dirname( self._fileName )
def __str__( self ):
instance = getattr( self, '_instance', None )
if instance is None:
return self.id()
else:
return self.id() + '_' + unicode( instance )
def addLink( self, destination, source, function=None ):
eNode = getattr( self, '_executionNode', None )
if eNode is None:
Parameterized.addLink( self, destination, source, function )
else:
eNode.addLink( destination, source, function )
def setExecutionNode( self, eNode ):
"""Sets the execution node of the pipeline.
:param eNode: object that describes the structure of the pipeline.
:type eNode: :py:class:`ExecutionNode`
"""
self._executionNode = eNode
def execution( self, context ):
"""
Execution function that is called when the process is run.
"""
if self._executionNode is not None:
return self._executionNode.run( context )
else:
raise RuntimeError( HTMLMessage(_t_( 'No execution method provided' )) )
def executionNode( self ):
"""Returns the execution node of the pipeline."""
return self._executionNode
def pipelineStructure( self ):
"""Returns the description of a pipeline in a dictionary or the id of the process if it is a simple process."""
return self.id()
def allProcesses( self ):
"""Returns the current process and all its children if it is a pipeline.
:rtype: generator
"""
yield self
if self._executionNode is not None:
stack = [ self._executionNode ]
while stack:
eNode = stack.pop( 0 )
if isinstance( eNode, ProcessExecutionNode ):
yield eNode._process
stack.extend( eNode.children() )
def allParameterFiles( self ):
"""Get recursively all parameters which are DiskItems, descending through
the pipeline structure.
"""
params = []
files = set()
for paramname in self.signature.iterkeys():
param = self.__dict__.get( paramname )
if param is not None and isinstance( param, DiskItem ):
filename = param.fullPath()
if filename not in files:
params.append( param )
files.add( filename )
# parse pipeline structure
eNodes = []
eNode = self.executionNode()
if eNode is not None:
eNodes.append( eNode )
while eNodes:
eNode = eNodes.pop()
eNodes += list( eNode.children() )
if isinstance( eNode, ProcessExecutionNode ):
process = eNode._process
for paramname in process.signature.iterkeys():
param = process.__dict__.get( paramname )
if param is not None and isinstance( param, DiskItem ):
filename = param.fullPath()
if filename not in files:
params.append( param )
files.add( filename )
return params
def saveStateInDictionary( self, result=None ):
"""Returns the description of the process in a dictionary."""
if result is None:
result = {}
result[ 'pipelineStructure' ] = self.pipelineStructure()
if self._executionNode is not None:
if self._executionNode._parameterized is not None:
Parameterized.saveStateInDictionary( self._executionNode._parameterized(), result )
eNodesState = {}
for eNodeKey in self._executionNode.childrenNames():
eNode = self._executionNode.child( eNodeKey )
eNodeDict = {}
eNode.saveStateInDictionary( eNodeDict )
eNodesState[ eNodeKey ] = eNodeDict
result[ 'executionNodes' ] = eNodesState
else:
Parameterized.saveStateInDictionary( self, result )
return result
def getAllParameters( self ):
"""
Returns all the parameters of the current process and its children if it is a pipeline.
:returns: tuples (Parameterized, attribute name, attribute type)
:rtype: generator
"""
stack = [ self ]
while stack:
node = stack.pop( 0 )
if isinstance( node, Process ):
parameterized = node
node = node.executionNode()
if node is not None:
stack += [node.child( i ) for i in node.childrenNames()]
else:
parameterized = node._parameterized
if parameterized is not None: parameterized = parameterized()
stack += [node.child( i ) for i in node.childrenNames()]
if parameterized is not None:
for attribute, type in parameterized.signature.iteritems():
yield ( parameterized, attribute, type )
#----------------------------------------------------------------------------
class IterationProcess( Process ):
"""
This class represents a set of process instances that can be executed in parallel.
It is used to iterate the same process on a set of data.
"""
def __init__( self, name, processes ):
self._id = name + 'Iteration'
self.name = name
self.instance = 1
self._processes = [getProcessInstance( p ) for p in processes]
Process.__init__( self )
for sp, p in zip( self._executionNode._children.values(), processes ):
if isinstance( p, ExecutionNode ):
sp._optional = p._optional
sp._selected = p._selected
#def __del__( self ):
#print 'del IterationProcess', self, ', children:', len( self._processes )
#print 'children:', self._processes
#del self._processes
#import gc
#print 'refs to execution node:', len( gc.get_referrers( self._executionNode ) )
#print [ i.keys() for i in gc.get_referrers( self._executionNode ) ]
def pipelineStructure( self ):
return { 'type': 'iteration', 'name' : self.name, 'children':[p.pipelineStructure() for p in self._processes] }
def initialization( self ):
if len( self._processes ) != 0:
dp = self._processes[0].name
else:
dp = None
eNode = ParallelExecutionNode( self.name, stopOnError=False,
possibleChildrenProcesses=dp, notify = True )
for i in xrange( len( self._processes ) ):
self._processes[ i ].isMainProcess = True
self._processes[ i ].name = repr(i+1) + ". " + self._processes[ i ].name
eNode.addChild( node = ProcessExecutionNode( self._processes[ i ],
optional = True, selected = True ) )
# Add callbacks to maintain synchronization
eNode.beforeChildAdded.add(\
ExecutionNode.MethodCallbackProxy( self.beforeChildAdded ) )
eNode.beforeChildRemoved.add(\
ExecutionNode.MethodCallbackProxy( self.beforeChildRemoved ) )
self._executionNode = eNode
def beforeChildAdded( self, parent, key, child ):
child._process.isMainProcess = True
child._process.name = repr(parent._internalIndex) + ". " + child._process.name
self._processes.append( child._process )
def beforeChildRemoved( self, parent, key, child ):
if child._process in self._processes:
self._processes.remove(child._process)
#----------------------------------------------------------------------------
class ListOfIterationProcess( IterationProcess ):
'''
An IterationProcess which has on its main signature a list of the first
element of each sub-process.
Used for viewers and editors of ListOf()
'''
class linkP( object ):
def __init__( self, proc, i ):
self.proc = proc
self.num = i
def __call__( self, par ):
if len( self.proc.param ) > self.num:
return self.proc.param[self.num]
def __init__( self, name, processes ):
IterationProcess.__init__( self, name, processes )
chs = list( self.executionNode().children() )[0]._process.signature
self.changeSignature( Signature( 'param', ListOf( chs.values()[0] ) ) )
en = self.executionNode()
en._parameterized = weakref.ref( self )
for i, p in enumerate( en.children() ):
s = p._process.signature
en.addLink( str(i) + '.' + s.keys()[0], 'param', self.linkP( self, i ) )
#----------------------------------------------------------------------------
class DistributedProcess( Process ):
"""
This class represents a set of process instances that can be executed in parallel.
"""
def __init__( self, name, processes ):
self._id = name + 'DistributedIteration'
self.name = name
self.instance = 1
self._processes = [getProcessInstance( p ) for p in processes]
Process.__init__( self )
for sp, p in zip( self._executionNode._children.values(), processes ):
if isinstance( p, ExecutionNode ):
sp._optional = p._optional
sp._selected = p._selected
def pipelineStructure( self ):
return { 'type': 'distributed', 'name' : self.name, 'children':[p.pipelineStructure() for p in self._processes] }
def initialization( self ):
eNode = ParallelExecutionNode( self.name )
for i in xrange( len( self._processes ) ):
self._processes[ i ].isMainProcess = True
subENode = self._processes[ i ]._executionNode
eNode.addChild( str( i ), ProcessExecutionNode( self._processes[ i ],
optional=True, selected = True ) )
self._executionNode = eNode
#----------------------------------------------------------------------------
class SelectionProcess( Process ):
"""
This class represents a choice between a list of processes.
"""
def __init__( self, name, processes ):
self._id = name + 'Selection'
self.name = name
self.instance = 1
self._processes = [getProcessInstance( p ) for p in processes]
Process.__init__( self )
for sp, p in zip( self._executionNode._children.values(), processes ):
if isinstance( p, ExecutionNode ):
sp._optional = p._optional
sp._selected = p._selected
def pipelineStructure( self ):
return { 'type': 'selection', 'name' : self.name, 'children':[p.pipelineStructure() for p in self._processes] }
def initialization( self ):
eNode = SelectionExecutionNode( self.name )
for i in xrange( len( self._processes ) ):
self._processes[ i ].isMainProcess = True
eNode.addChild( str( i ), ProcessExecutionNode( self._processes[ i ],
optional=True, selected = True ) )
self._executionNode = eNode
#----------------------------------------------------------------------------
class TimeoutCall( object ):
def __init__( self, function, *args, **kwargs ):
self.function = function
self.args = args
self.kwargs = kwargs
self.event = threading.Event()
self.callFunction = 0
self.functionLock =threading.RLock()
def __del__( self ):
self.stop()
def _thread( self ):
self.event.wait( self.timeout )
self.functionLock.acquire()
try:
if self.callFunction:
apply( self.function, self.args, self.kwargs )
finally:
self.functionLock.release()
def start( self, timeout ):
self.stop() # Just in case of multiple start() call during timeout
self.functionLock.acquire()
try:
self.callFunction = 1
self.event.clear()
self.timeout = timeout
threading.Thread( target = self._thread )
finally:
self.functionLock.release()
def stop( self ):
self.functionLock.acquire()
try:
self.callFunction = 0
self.event.set()
finally:
self.functionLock.release()
#----------------------------------------------------------------------------
def signalName( signalNumber ):
for key, value in signal.__dict__.items():
if key[ :3 ] == 'SIG' and value == signalNumber:
return key
return str( signalNumber )
#----------------------------------------------------------------------------
def escapeQuoteForShell( s ):
return string.replace( s, "'", "'\"'\"'" )
#-------------------------------------------------------------------------------
class ExecutionNode( object ):
"""
Base class for the classes that describe a pipeline structure.
"""
class MultiParameterLink:
def __init__( self, sources, function ):
self.sources = []
for p, n in sources:
if type(p) is weakref.ReferenceType:
self.sources.append( ( p, n ) )
else:
self.sources.append( ( weakref.ref( p ), n ) )
self.function = function
self.hasParameterized = hasParameter( function, 'parameterized' )
self.hasNames = hasParameter( function, 'names' )
def __call__( self, dummy1, dummy2 ):
kwargs = {}
if self.hasParameterized:
kwargs[ 'parameterized' ] = [i[0]() for i in self.sources]
if self.hasNames:
kwargs[ 'names' ] = [i[1] for i in self.sources]
return self.function( *[getattr( i[0](), i[1], None ) for i in self.sources],
**kwargs )
class MethodCallbackProxy( object ):
def __init__( self, method ):
self.object = weakref.ref( method.im_self )
self.method = method.im_func
def __call__( self, *args, **kwargs ):
o = self.object()
if o is not None:
self.method( o, *args, **kwargs )
def __eq__( self, other ):
if isinstance( other, ExecutionNode.MethodCallbackProxy ):
return self.object() == other.object() and self.method == other.method
if self.object() is None:
return other is None
return self.method.__get__( self.object() ) == other
def __init__( self, name='', optional = False, selected = True,
guiOnly = False, parameterized = None, expandedInGui = False ):
"""
:param string name: name of the node - default ''.
:param boolean optional: indicates if this node is optional in the pipeline - default False.
:param boolean selected: indicates if the node is selected in the pipeline - default True.
:param boolean guiOnly: default False.
:param parameterized: :py:class:`Parameterized` containing the signature of the node - default None.
"""
# Initialize an empty execution node
#print 'ExecutionNode.__init__', self
self.__dict__[ '_deleted' ] = False # safety to avoid double deletion
# see http://code.activestate.com/lists/python-list/191512/
self.__dict__[ '_children' ] = SortedDictionary()
if parameterized is not None:
parameterized = weakref.ref( parameterized )
self.__dict__[ '_parameterized' ] = parameterized
self.__dict__[ '_name' ] = str( name )
self.__dict__[ '_optional' ] = optional
self.__dict__[ '_selected' ] = selected
self.__dict__[ '_guiOnly' ] = guiOnly
self.__dict__[ '_selectionChange' ] = Notifier( 1 )
self.__dict__[ '_expandedInGui' ] = expandedInGui
self.__dict__[ '_dependencies' ] = []
def __del__( self ):
#print 'del ExecutionNode', self
if not hasattr( self, '_deleted' ) or self.__dict__[ '_deleted' ]:
# print '*** ExecutionNode already deleted ! ***'
return
self.__dict__[ '_deleted' ] = True
debugHere()
def _copy(self, node, withparams=True):
"""
Uses non default parameters values to initialize the parameters of the node given in argument, if withparams is True (which is the default).
"""
# if execution node contains a process, copy the process parameters and copy its execution node parameters if any
process=getattr(self, "_process", None)
if process:
processCopy=node._process
if withparams:
for ( n, v ) in process.signature.items():
#if not self.isDefault( n ):
#processCopy.setValue( n, getattr( process, n, None ),
#default=False )
processCopy.setValue( n, getattr( process, n, None ),
default=process.isDefault( n ) )
processNode=process.executionNode()
if processNode:
processNode._copy(processCopy.executionNode(), withparams=withparams)
node.setSelected(self._selected)
# if execution node have children nodes, copy the parameters of these nodes
for name in self.childrenNames():
child=self.child(name)
child._copy(node.child(name), withparams=withparams)
def addChild( self, name, node, index = None):
'''Add a new child execution node.
:param string name: name which identifies the node
:param node: an :py:class:`ExecutionNode` which will be added to this node's children.
'''
if self._children.has_key( name ):
raise KeyError( HTMLMessage(_t_( '%s already defined' ) % ( name, )) )
if not isinstance( node, ExecutionNode ):
raise RuntimeError( HTMLMessage('node argument must be an execution node') )
if not index is None :
self._children.insert( index, name, node )
else :
self._children[ name ] = node
def removeChild( self, name ):
'''Remove child execution node.
:param string name: name which identifies the node
'''
if not self._children.has_key( name ):
raise KeyError( HTMLMessage(_t_( '%s not defined' ) % ( name, )) )
c = self._children[ name ]
del self._children[ name ]
return c
def childrenNames( self ):
'''
Returns the list of names of the children execution nodes.
'''
return self._children.keys()
def children( self ):
"""Returns the list of children execution nodes."""
return self._children.itervalues()
def hasChildren( self ):
"""Returns True if this node has children."""
return bool( self._children )
def setSelected( self, selected ):
"""Change the selection state of the node.
:param bool selected: new selection state of the node. If the selection changes, a selection change notifier is notified.
"""
if selected != self._selected:
self._selected = selected
self._selectionChange.notify( self )
def isSelected( self ):
"""True if this node is selected."""
return self._selected
def __setattr__( self, attribute, value ):
"""
If the attribute is in the signature of the corresponding parameterized object, it is modified.
"""
if self._parameterized is not None and \
self._parameterized().signature.has_key( attribute ):
setattr( self._parameterized(), attribute, value )
elif self._children.has_key( attribute ):
raise RuntimeError( HTMLMessage(_t_( 'Direct modification of execution node %s is not allowed.' ) % ( attribute, )) )
else:
self.__dict__[ attribute ] = value
def __getattr__( self, attribute ):
p = self.__dict__.get( '_parameterized' )
if p is not None: p = p()
if p is not None and hasattr( p, attribute ):
return getattr( p, attribute )
children = self.__dict__[ '_children' ]
if children.has_key( attribute ):
return children[ attribute ]
raise AttributeError( attribute )
def child( self, name, default = None ):
"""Get a child node by name."""
return self._children.get( name, default )
def allParameterFiles( self ):
"""Get recursively all parameters which are DiskItems, descending through
the pipeline structure.
"""
params = []
files = set()
# parse pipeline structure
eNodes = [ self ]
while eNodes:
eNode = eNodes.pop()
eNodes += list( eNode.children() )
if isinstance( eNode, ProcessExecutionNode ):
process = eNode._process
for paramname in process.signature.iterkeys():
param = process.__dict__.get( paramname )
if param is not None and isinstance( param, DiskItem ):
filename = param.fullPath()
if filename not in files:
params.append( param )
files.add( filename )
return params
def run( self, context ):
"""
Calls :py:meth:`_run` method if the node is selected.
"""
if self._optional and ( not self._selected ):
context.write( 'Skip unselected node: ' + str(self.name()) + '' )
return
if self._guiOnly and not neuroConfig.gui:
context.write( 'Skip GUI-only node: ' + str(self.name()) + '' )
return
return self._run( context )
def _run( self, context ):
"""
Does nothing in the base class. It is overriden by derived classes.
"""
pass
def name( self ):
"""Returns the name of the node."""
return self._name
def gui( self, parent, processView = None ):
"""
Returns the graphical user interface of this node.
"""
from brainvisa.processing.qtgui.neuroProcessesGUI import ExecutionNodeGUI
if self._parameterized is not None:
if processView != None and processView.read_only:
return ExecutionNodeGUI(parent, self._parameterized(), read_only=True)
return ExecutionNodeGUI(parent, self._parameterized())
return None
def addLink( self, destination, source, function=None ):
"""
Adds a parameter link like :py:meth:`Parameterized.addLink`.
"""
# Parse source
sources = []
if type( source ) in ( types.ListType, types.TupleType ):
for i in source:
sources.append( self.parseParameterString( i ) )
else:
sources.append( self.parseParameterString( source ) )
destObject, destParameter = self.parseParameterString( destination )
if destObject is not None:
destObject = weakref.proxy( destObject )
# Check if a default function can be provided
if function is None:
if len( sources ) == 1:
function = lambda x: x
else:
raise RuntimeError( HTMLMessage(_t_( 'No function provided in addLink' )) )
multiLink = self.MultiParameterLink( sources, function )
for sourceObject, sourceParameter in sources:
sourceObject._links.setdefault( sourceParameter, [] ).append (
( destObject, destParameter, multiLink, True ) )
def addDoubleLink( self, destination, source, function=None ):
"""
Creates a double link source -> destination and destination -> source.
"""
self.addLink( destination, source, function )
self.addLink( source, destination, function )
def removeLink( self, destination, source, function=None ):
"""
Removes a parameters link added with :py:meth:`addLink`.
"""
# Parse sourceExecutionContext
sources = []
if type( source ) in ( types.ListType, types.TupleType ):
for i in source:
sources.append( self.parseParameterString( i ) )
else:
sources.append( self.parseParameterString( source ) )
destObject, destParameter = self.parseParameterString( destination )
removed = 0
for sourceObject, sourceParameter in sources:
l = sourceObject._links.get( sourceParameter, [] )
if l:
lbis = l
l = [i for i in l if ( destObject and i[0] is not destObject and ( i[0] is not weakref.proxy( destObject ) ) ) or i[1] != destParameter]
if len(l) != len(lbis):
removed = 1
if l:
sourceObject._links[ sourceParameter ] = l
else:
del sourceObject._links[ sourceParameter ]
removed=1
if removed == 0:
print 'warning: enode link not removed:', self, destination, 'from:', source, ', function:', function
def removeDoubleLink( self, destination, source, function=None ):
"""
Removes a double link source -> destination and destination -> source.
"""
self.removeLink( destination, source, function )
self.removeLink( source, destination, function )
def parseParameterString( self, parameterString ):
"""
Returns a tuple containing the :py:class:`Parameterized` object of the child node indicated in the parameter string and the name of the parameter.
:param string parameterString: references a parameter of a child node with a path like ....
"""
if parameterString is None: return ( None, None )
l = parameterString.split( '.' )
node = self
for nodeName in l[ : -1 ]:
node = node.child( nodeName )
parameterized = node._parameterized
if parameterized is not None: parameterized = parameterized()
name = l[ -1 ]
if parameterized is None or not parameterized.signature.has_key( name ):
raise KeyError( name )
return ( parameterized, name )
def saveStateInDictionary( self, result=None ):
if result is None:
result = {}
result[ 'name' ] = self._name
result[ 'selected' ] = self._selected
if self._parameterized is not None:
Parameterized.saveStateInDictionary( self._parameterized(), result )
eNodesState = {}
for eNodeKey in self.childrenNames():
eNode = self.child( eNodeKey )
eNodesState[ eNodeKey ] = eNode.saveStateInDictionary()
result[ 'executionNodes' ] = eNodesState
return result
def addExecutionDependencies( self, deps ):
'''Adds to the execution node dependencies on the execution of other nodes.
This allows to build a dependencies structure which is not forced to be a
tree, but can be a grapĥ. Dependencies are used to build Soma-Workflow
workflows with correct dependencies.
'''
if type( deps ) not in ( types.ListType, types.TupleType ):
deps = [ deps ]
self._dependencies += [ weakref.ref(x) for x in deps ]
#-------------------------------------------------------------------------------
class ProcessExecutionNode( ExecutionNode ):
'''
An execution node that has no children and run one process
'''
def __init__( self, process, optional = False, selected = True,
guiOnly = False, expandedInGui = False, altname = None ):
process = getProcessInstance( process )
#print 'ProcessExecutionNode.__init__:', self, process.name
ExecutionNode.__init__( self, process.name,
optional = optional,
selected = selected,
guiOnly = guiOnly,
parameterized = process,
expandedInGui = expandedInGui )
self.__dict__[ '_process' ] = process
if altname is not None:
self.__dict__[ '_name' ] = altname
reloadNotifier = getattr( process, 'processReloadNotifier', None )
if reloadNotifier is not None:
reloadNotifier.add( ExecutionNode.MethodCallbackProxy( \
self.processReloaded ) )
def __del__( self ):
#print 'del ProcessExecutionNode', self
if not hasattr( self, '_deleted' ) or self._deleted:
# print '*** already deleted !***'
return
if hasattr( self, '_process' ):
#print ' del proc:', self._process.name
reloadNotifier = getattr( self._process, 'processReloadNotifier', None )
if reloadNotifier is not None:
try:
l = len( reloadNotifier._listeners )
z = ExecutionNode.MethodCallbackProxy( self.processReloaded )
# bidouille: hack z so as to contain a weakref to None
# since we are in __del__ and existing weakrefs to self have already
# been neutralized
class A(object): pass
w = weakref.ref(A()) # w points to None immediately
z.object = w
x = reloadNotifier.remove( z )
except AttributeError:
# this try..except is here to prevent an error when quitting
# BrainVisa:
# ProcessExecutionNode class is set to None during module destruction
pass
else:
# print 'del ProcessExecutionNode', self
print 'no _process in ProcessExecutionNode !'
try:
ExecutionNode.__del__( self )
except:
# same as above
pass
def addChild( self, name, node, index = None ):
raise RuntimeError( _t_( 'A ProcessExecutionNode cannot have children' ) )
def _run( self, context ):
return context.runProcess( self._process )
def gui( self, parent, processView = None ):
if processView is not None:
return ProcessView( self._process, parent,
externalInfo = processView.info,
read_only=processView.read_only)
else:
return ProcessView( self._process, parent )
def children( self ):
eNode = getattr( self._process, '_executionNode', None )
if eNode is not None:
return eNode._children.itervalues()
else:
return []
def childrenNames( self ):
eNode = getattr( self._process, '_executionNode', None )
if eNode is not None:
return eNode._children.keys()
else:
return []
def __setattr__( self, attribute, value ):
if self._parameterized is not None and \
self._parameterized().signature.has_key( attribute ):
setattr( self._parameterized(), attribute, value )
else:
eNode = getattr( self._process, '_executionNode', None )
if eNode is not None and eNode._children.has_key( attribute ):
raise RuntimeError( HTMLMessage(_t_( 'Direct modification of execution node %s is not allowed.' ) % ( attribute, )) )
self.__dict__[ attribute ] = value
def __getattr__( self, attribute ):
p = self.__dict__.get( '_parameterized' )()
if p is not None and hasattr( p, attribute ):
return getattr( p, attribute )
eNode = getattr( self._process, '_executionNode', None )
if eNode is not None:
c = eNode.child( attribute )
if c is not None:
return c
raise AttributeError( attribute )
def child( self, name, default=None ):
eNode = getattr( self._process, '_executionNode', None )
if eNode is not None:
return eNode.child( name, default )
return default
def processReloaded( self, newProcess ):
"""
If the associated process has an attribute *processReloadNotifier*, this callback is attached to the notifier.
So, the node is reloaded when the process is reloaded.
"""
event = ProcessExecutionEvent()
event.setProcess( self._process )
self._process.processReloadNotifier.remove( ExecutionNode.MethodCallbackProxy( self.processReloaded ) )
self.__dict__[ '_process' ] = getProcessInstanceFromProcessEvent( event )
self._process.processReloadNotifier.add( ExecutionNode.MethodCallbackProxy( self.processReloaded ) )
def addExecutionDependencies( self, deps ):
ExecutionNode.addExecutionDependencies( self, deps )
eNode = self._process._executionNode
if eNode:
eNode.addExecutionDependencies( deps )
#-------------------------------------------------------------------------------
class SerialExecutionNode( ExecutionNode ):
'''An execution node that run all its children sequentially'''
def __init__(self, name='', optional = False, selected = True,
guiOnly = False, parameterized = None, stopOnError=True,
expandedInGui = False, possibleChildrenProcesses = None, notify = False ):
#print 'SerialExecutionNode.__init__', self
ExecutionNode.__init__(self, name, optional, selected, guiOnly, parameterized, expandedInGui=expandedInGui )
self.stopOnError=stopOnError
self.notify = notify
if possibleChildrenProcesses :
if not isinstance( possibleChildrenProcesses, dict ) :
if not isinstance( possibleChildrenProcesses, list ) \
and not isinstance( possibleChildrenProcesses, tuple ) :
possibleChildrenProcesses = [ possibleChildrenProcesses ]
r = {}
for i in xrange(len(possibleChildrenProcesses)) :
r[possibleChildrenProcesses[i]] = { 'optional' : True,
'selected' : True,
'expandedInGui' : False }
possibleChildrenProcesses = r
self._internalIndex = 0
self.possibleChildrenProcesses = possibleChildrenProcesses
if self.notify :
# Add child changes notifiers
self.beforeChildRemoved = Notifier( 4 )
self.afterChildRemoved = Notifier( 4 )
self.beforeChildAdded = Notifier( 4 )
self.afterChildAdded = Notifier( 4 )
def _run( self, context ):
result = []
pi, p = context.getProgressInfo( self )
pi.children = [ None ] * len( self._children )
if self.stopOnError:
for node in self._children.values():
npi, proc = context.getProgressInfo( node, parent=pi )
context.progress()
result.append( node.run( context ) )
del npi
else:
for node in self._children.values():
npi, proc = context.getProgressInfo( node, parent=pi )
context.progress()
try:
result.append( node.run( context ) )
del npi
except ExecutionContext.UserInterruptionStep, e:
context.error(unicode(e))
except ExecutionContext.UserInterruption:
raise
except Exception, e:
context.error("Error in execution node : "+unicode(e))
context.progress()
return result
def addChild(self, name = None, node = None, index = None):
if self.possibleChildrenProcesses :
if not name :
if isinstance( node, ExecutionNode ):
name = node.name() + '_' + str(self._internalIndex)
else :
raise RuntimeError( HTMLMessage('node argument must be an execution node') )
else:
name += '_' + str(self._internalIndex)
if not node:
node = self.possibleChildrenProcesses
self._internalIndex += 1
if self.notify :
self.beforeChildAdded.notify( weakref.proxy( self ), name, weakref.proxy( node ) )
super( SerialExecutionNode, self ).addChild(name, node, index)
if self.notify :
self.afterChildAdded.notify( weakref.proxy( self ), name, weakref.proxy( node ) )
def removeChild(self, name):
if self.possibleChildrenProcesses :
if not self._children.has_key( name ):
raise KeyError( HTMLMessage(_t_( '%s not defined' ) % ( name, )) )
if self.notify :
node = self._children[ name ]
self.beforeChildRemoved.notify( weakref.proxy( self ), name, weakref.proxy( node ) )
super( SerialExecutionNode, self ).removeChild(name)
if self.notify :
self.afterChildRemoved.notify( weakref.proxy( self ), name, weakref.proxy( node ) )
#-------------------------------------------------------------------------------
class ParallelExecutionNode( SerialExecutionNode ):
"""
An execution node that run all its children in any order (and in parallel
if possible)
"""
def _run( self, context ):
pi, p = context.getProgressInfo( self )
# do as for serial node
return super( ParallelExecutionNode, self )._run( context )
#-------------------------------------------------------------------------------
class SelectionExecutionNode( ExecutionNode ):
'''An execution node that run one of its children'''
def __init__( self, *args, **kwargs ):
#print 'SelectionExecutionNode.__init__', self
ExecutionNode.__init__( self, *args, **kwargs )
def __del__( self ):
#print 'SelectionExecutionNode.__del__', self
if not hasattr( self, '_deleted' ) or self._deleted:
# print '*** SelectionExecutionNode already deleted'
return
for node in self._children.values():
node._selectionChange.remove( ExecutionNode.MethodCallbackProxy( \
self.childSelectionChange ) )
#print '__del__ finished'
def _run( self, context ):
'Run the selected child'
if self._selected is None:
raise RuntimeError( _t_( 'No children selected' ) )
pi, p = context.getProgressInfo( self )
pi.children = [ None ]
for node in self._children.values():
if node._selected:
npi, proc = context.getProgressInfo( node, parent=pi )
context.progress()
res = node.run( context )
del npi
context.progress()
return res
context.progress()
def addChild( self, name, node, index = None ):
'Add a new child execution node'
ExecutionNode.addChild(self, name, node, index)
node._selectionChange.add( ExecutionNode.MethodCallbackProxy( \
self.childSelectionChange ) )
node._dependencies += self._dependencies
def childSelectionChange(self, node):
'''This callback is called when the selection state of a child has changed.
If the child is selected, all the other children must be unselected
because this node is a selectionNode.'''
if node._selected:
for child in self.children():
if child != node:
child.setSelected(False)
def addExecutionDependencies( self, deps ):
ExecutionNode.addExecutionDependencies( self, deps )
for node in self._children.values():
node.addExecutionDependencies( deps )
#-------------------------------------------------------------------------------
class ExecutionContext( object ):
"""
This object represents the execution context of the processes.
Indeed, a process can be started in different contexts :
* The user starts the process by clicking on the Run button in the graphical interface.
* The process is started via a script. It is possible to run brainvisa in batch mode (without any graphical interface) and to run a process via a python function : brainvisa.processes.defaultContext().runProcess(...).
* The process is a converter, so it can be run automatically by BrainVISA when a conversion is needed for another process parameters.
* The process is a viewer or an editor, it is run when the user clicks on the corresponding icon to view or edit another process parameter.
The interactions with the user are different according to the context. That's why the context object offers several useful functions to interact with BrainVISA and to call system commands.
Here are these functions :
* :py:meth:`write`, :py:meth:`warning`, :py:meth:`error` : prints a message, either in the graphical process window (in GUI mode) or in the terminal (in batch mode).
* :py:meth:`log` : writes a message in the BrainVISA log file.
* :py:meth:`ask`, :py:meth:`dialog` : asks a question to the user.
* :py:meth:`temporary` : creates a temporary file.
* :py:meth:`system`: calls a system command.
* :py:meth:`runProcess` : runs a BrainVISA process.
* :py:meth:`checkInterruption` : defines a breakpoint.
"""
remote = None
class UserInterruption( Exception ):
def __init__( self ):
Exception.__init__( self, _t_( 'user interruption' ) )
class UserInterruptionStep( Exception ):
def __init__( self ):
Exception.__init__( self, _t_( 'user interruption of current step' ) )
class StackInfo:
def __init__( self, process ):
self.process = process
self.processCount = {}
self.thread = None
self.debug = None
self.log = None
self.time = time.localtime()
def __init__( self, userLevel = None, debug = None ):
if userLevel is None:
self.userLevel = neuroConfig.userLevel
else:
self.userLevel = userLevel
#self._processStack = []
self._lock = threading.RLock()
self._processStackThread = {}
self._processStackHead = None
self.manageExceptions = 1
self._systemOutputLevel = 0
self._systemLog = None
self._systemLogFile = None
self._interruptionRequest = None
self._interruptionActions = {}
self._interruptionActionsId = 0
self._interruptionLock = threading.RLock()
self._allowHistory = True
def _processStack( self ):
self._lock.acquire()
try:
stack = self._processStackThread[ threading.currentThread() ]
except:
stack = []
self._processStackThread[ threading.currentThread() ] = stack
self._lock.release()
return stack
def _popStack( self ):
self._lock.acquire()
stack = self._processStackThread[ threading.currentThread() ]
stackinfo = stack.pop()
if len( stack ) == 0:
del self._processStackThread[ threading.currentThread() ]
if stackinfo is self._processStackHead:
self._processStackHead = None
self._lock.release()
return stackinfo
def _pushStack( self, stackinfo ):
self._lock.acquire()
stack = self._processStack()
stack.append( stackinfo )
if self._processStackHead is None:
self._processStackHead = stackinfo
self._lock.release()
def _stackTop( self ):
stack = self._processStack()
if len( stack ) == 0:
return None
return stack[-1]
def _processStackParent( self ):
stack = self._processStack()
if len( stack ) == 0:
return self._processStackHead
return stack[-1]
def _setArguments( self, _process, *args, **kwargs ):
# Set arguments
for i, v in enumerate( args ):
n = _process.signature.keys()[ i ]
_process._setImmutable( n, True )
# performing this 2 pass loop allows to set parameters with
# a forced value to immutable (ie non-linked) before actually
# setting values and running links. This avoids a bunch of unnecessary
# links to work (often several times)
for ( n, v ) in kwargs.items():
_process._setImmutable( n, True )
for i, v in enumerate( args ):
n = _process.signature.keys()[ i ]
_process.setDefault( n, 0 )
if v is not None:
_process.setValue( n, v )
else:
setattr( _process, n, None )
for ( n, v ) in kwargs.items():
_process.setDefault( n, 0 )
if v is not None:
_process.setValue( n, v )
else:
setattr( _process, n, None )
_process._clearImmutableParameters()
_process.checkArguments()
def _startProcess( self, _process, executionFunction, *args, **kwargs ):
if not isinstance( _process, Process ):
_process = getProcessInstance( _process )
apply( self._setArguments, (_process,)+args, kwargs )
# Launch process
t = threading.Thread( target = self._processExecutionThread,
args = ( _process, executionFunction ) )
t.start()
return _process
def runProcess( self, _process, *args, **kwargs ):
"""
It is possible to call a sub-process in the current process by calling context.runProcess.
The first argument is the process identifier, which is either the filename wihtout extension of the process or its english name.
The other arguments are the values of the process parameters. All mandatory argument must have a value.
The function returns the value returned by the sub-process execution method.
*Example*
>>> context.runProcess( 'do_something', self.input, self.output, value = 3.14 )
In this example, the process do_something is called with self.input as the first paramter value, self.ouput as the second parameter value and 3.14 to the parameter named value.
"""
_process = getProcessInstance( _process )
self.checkInterruption()
apply( self._setArguments, (_process,)+args, kwargs )
result = self._processExecution( _process, None )
self.checkInterruption()
if self._lastProcessRaisedException:
e = self._lastProcessRaisedException
self._lastProcessRaisedException = None # reset exception once used.
if isinstance( e, tuple ) and len( e ) == 2:
# e = ( exception, traceback )
raise e[0], None, e[1]
else:
raise e
return result
@staticmethod
def createContext():
return ExecutionContext()
def runInteractiveProcess( self, callMeAtTheEnd, process, *args, **kwargs ):
"""
Runs a process in a new thread and calls a callback function when the execution is finished.
:param function callMeAtTheEnd: callback function which will be called the process execution is finished.
:param process: id of the process which will be run.
"""
context = self.createContext()
process = getProcessInstance( process )
self.checkInterruption()
apply( self._setArguments, (process,)+args, kwargs )
thread = threading.Thread( target = self._runInteractiveProcessThread,
args = ( context, process, callMeAtTheEnd ) )
thread.start()
def _runInteractiveProcessThread( self, context, process, callMeAtTheEnd ):
try:
result = context.runProcess( process )
except Exception, e:
result = e
callMeAtTheEnd( result )
def _processExecutionThread( self, *args, **kwargs ):
self._processExecution( *args, **kwargs )
neuroHierarchy.databases.currentThreadCleanup()
def _processExecution( self, process, executionFunction=None ):
'''Execute the process "process". The value return is stored to avoid
the garbage-collection of some of the objects created by the process
itself (GUI for example).
'''
result = None
stackTop = None
process = getProcessInstance( process )
stack = self._processStack()
stackTop = self._processStackParent()
if stackTop:
## if neuroConfig.userLevel > 0:
## self.write( ' + ')
' \
## + _t_(process.name) + ' '\
## + str(process.instance) + '' )
# Count process execution
count = stackTop.processCount.get( process._id, 0 )
stackTop.processCount[ process._id ] = count + 1
newStackTop = self.StackInfo( process )
self._pushStack( newStackTop )
ishead = not stackTop
# Logging process start
if not stackTop:
process.isMainProcess = True
try: # finally -> processFinished
try: # show exception
# check write parameters if the process is the main process (check all parameters in child nodes if it is a pipeline)
# or if it has a parent which is not a pipeline that is to say, the current process is run throught context.runProcess
if ishead:
self._allWriteDiskItems = {}
if ishead or (stackTop and stackTop.process._executionNode is None):
writeParameters = []
for parameterized, attribute, type in process.getAllParameters():
if isinstance( type, WriteDiskItem ):
item = getattr( parameterized, attribute )
if item is not None:
writeParameters.append(item)
elif isinstance( type, ListOf ) and isinstance( type.contentType, WriteDiskItem ):
itemList = getattr( parameterized, attribute )
if itemList:
writeParameters.extend(itemList)
for item in writeParameters:
dirname = os.path.dirname( item.fullPath() )
if not os.path.exists( dirname ):
safemkdir.makedirs( dirname )
uuid=item.uuid()
self._allWriteDiskItems[uuid] = [ item, item.modificationHash() ]
if ishead:
log = neuroConfig.mainLog
else:
if len( stack ) >= 2:
log = stack[ -2 ].log
else:
# FIXME:
# attaching to head log is not always the right solution
# if a sub-process has parallel sub-nodes, then a new thread
# and a new stack will be created, but the logs will not be
# appended to the correct parent
log = self._processStackHead.log
if log is not None:
#print "Create subLog for process ", process.name
newStackTop.log = log.subLog()
process._log = newStackTop.log
content = '
' + _t_(process.name) + '
' + _t_('Process identifier') + '
' + process._id
content += '' + _t_('Execution platform') +'
'
if(hasattr(process, 'executionWorkflow')):
content += 'Soma-Workflow'
else :
content += 'BrainVISA'
content += '' + _t_('Parameters') +'
'
for n in process.signature.keys():
content += '' + n + ' = ' + htmlEscape( str( getattr( process, n, None ) ) ) + ''
content += '
' + _t_( 'Output' ) + '
'
try:
#print "Create subTextLog for process ", process.name
process._outputLog = log.subTextLog()
process._outputLogFile = open( process._outputLog.fileName, 'w' )
print >> process._outputLogFile, content
process._outputLogFile.flush()
content = process._outputLog
except:
content += '' + _t_('Unabled to open log file') + ''
process._outputLog = None
process._outputLogFile = None
if stackTop:
self._lastStartProcessLogItem = log.append( _t_(process.name), html=content,
children=newStackTop.log, icon='icon_process.png' )
else:
self._lastStartProcessLogItem = log.append( _t_(process.name) + ' ' + str(process.instance), html=content,
children=newStackTop.log, icon='icon_process.png' )
else:
newStackTop.log = None
if ishead and self._allowHistory:
self._historyBookEvent, self._historyBooksContext = HistoryBook.storeProcessStart( self, process )
self._processStarted()
newStackTop.thread = threading.currentThread()
self._lastProcessRaisedException = None
# Check arguments and conversions
def _getConvertedValue( v, p ):
# v: value
# p: parameter (Read/WriteDiskItem)
if v and getattr(v, "type", None) and ( ( not isSameDiskItemType( v.type, p.type ) ) or v.format not in p.formats ):
c = None
formats = [ p.preferredFormat ] \
+ [ f for f in p.formats if f is not p.preferredFormat ]
for destinationFormat in formats:
converter = getConverter( (v.type, v.format), (p.type, destinationFormat), checkUpdate=False )
if converter:
tmp = self.temporary( destinationFormat )
tmp.type = v.type
tmp.copyAttributes( v )
convargs = { 'read' : v, 'write' : tmp }
c = getProcessInstance( converter.name )
if c is not None:
try:
apply( self._setArguments, (c,), convargs )
if c.write is not None:
break
except:
pass
## if not converter: raise Exception( _t_('Cannot convert format %s to format %s for parameter %s') % ( _t_( v.format.name ), _t_( destinationFormat.name ), n ) )
## tmp = self.temporary( destinationFormat )
## tmp.type = v.type
## tmp.copyAttributes( v )
## self.runProcess( converter.name, read = v, write = tmp )
if not c: raise Exception( HTMLMessage(_t_('Cannot convert format %s to format %s for parameter %s') % ( _t_( v.format.name ), _t_( destinationFormat.name ), n )) )
self.runProcess( c )
return tmp
converter = None
for ( n, p ) in process.signature.items():
if isinstance( p, ReadDiskItem ) and p.enableConversion:
v = getattr( process, n )
tmp = _getConvertedValue( v, p )
if tmp is not None:
process.setConvertedValue( n, tmp )
elif isinstance( p, WriteDiskItem ):
v = getattr( process, n )
#test if data is locked
if (v is not None ):
if v.isLockData() and (process.execution.im_func != super(process.__class__, process).execution.im_func ) :
# raise an error if the diskitem is an output of a process which has an execution function (not the default execution function of the Process class)
raise IOError ( HTMLMessage(_t_('The file: %s is locked. It cannot be opened for writing. You can unlock it if necessary using the contextual menu of the parameter %s') % ( str(v), n ) ))
#end test if data is locked
if v is not None:
v.createParentDirectory()
elif isinstance( p, ListOf ):
needsconv = False
converted = []
lv = getattr( process, n )
for v in lv:
tmp = _getConvertedValue( v, p.contentType )
if tmp is not None:
converted.append( tmp )
needsconv = True
else:
converted.append( v )
if needsconv:
process.setConvertedValue( n, converted )
if executionFunction is None:
if(hasattr(process, 'executionWorkflow')) :
from soma_workflow.client import WorkflowController, Workflow, Helper
jobs, dependencies, root_group = process.executionWorkflow( self )
workflow = Workflow(jobs = jobs, dependencies = dependencies, root_group = root_group)
controller = WorkflowController()
wid = controller.submit_workflow(workflow=workflow, name=process.name)
Helper.wait_workflow(wid, controller)
list_failed_jobs = Helper.list_failed_jobs(wid, controller)
result = workflow
if (len(list_failed_jobs) > 0):
raise Exception('run through soma workflow failed, see details with soma-workflow-gui')
else :
# Delete the submitted workflow
controller.delete_workflow(wid, True)
else :
result = process.execution( self )
else:
result = executionFunction( self )
except Exception, e:
self._lastProcessRaisedException = ( e, sys.exc_info()[2] )
try:
self._showException()
except SystemExit, e:
neuroConfig.exitValue = e.args[0]
except:
import traceback
info = sys.exc_info()
sys.stderr.write('\n%s: %s\n' % (info[0].__name__, unicode(info[1])))
traceback.print_tb(info[2], None, sys.stderr)
logException( context=self )
if self._depth() != 1 or not self.manageExceptions:
raise
finally:
self._processFinished( result )
process.restoreConvertedValues()
# update history
if self._allowHistory and ishead and hasattr( self, '_historyBookEvent' ) and self._historyBookEvent is not None: # one of these conditions may be false when an exception occurs during execution
HistoryBook.storeProcessFinished( self, process, self._historyBookEvent, self._historyBooksContext )
self._historyBookEvent = None
self._historyBooksContext = None
for item_hash in self._allWriteDiskItems.values():
item, hash = item_hash
if item.isReadable():
if item.modificationHash() != hash:
try:
# do not try to insert in the database an item that doesn't have any reference to a database
# or which is temporary
if item.get("_database", None) and \
( not hasattr( item, '_isTemporary' ) \
or not item._isTemporary ):
neuroHierarchy.databases.insertDiskItem( item, update=True )
except NotInDatabaseError:
pass
except:
showException()
item_hash[ 1 ] = item.modificationHash()
elif (process.isMainProcess): # clear unused minfs only when the main process is finished to avoid clearing minf that will be used in next steps
item.clearMinf()
# Close output log file
if process._outputLogFile is not None:
print >> process._outputLogFile, '
' \
## + _t_(process.name) + ' '\
## + str(process.instance) + '' ) # Count process execution count = stackTop.processCount.get( process._id, 0 ) stackTop.processCount[ process._id ] = count + 1 newStackTop = self.StackInfo( process ) self._pushStack( newStackTop ) ishead = not stackTop # Logging process start if not stackTop: process.isMainProcess = True try: # finally -> processFinished try: # show exception # check write parameters if the process is the main process (check all parameters in child nodes if it is a pipeline) # or if it has a parent which is not a pipeline that is to say, the current process is run throught context.runProcess if ishead: self._allWriteDiskItems = {} if ishead or (stackTop and stackTop.process._executionNode is None): writeParameters = [] for parameterized, attribute, type in process.getAllParameters(): if isinstance( type, WriteDiskItem ): item = getattr( parameterized, attribute ) if item is not None: writeParameters.append(item) elif isinstance( type, ListOf ) and isinstance( type.contentType, WriteDiskItem ): itemList = getattr( parameterized, attribute ) if itemList: writeParameters.extend(itemList) for item in writeParameters: dirname = os.path.dirname( item.fullPath() ) if not os.path.exists( dirname ): safemkdir.makedirs( dirname ) uuid=item.uuid() self._allWriteDiskItems[uuid] = [ item, item.modificationHash() ] if ishead: log = neuroConfig.mainLog else: if len( stack ) >= 2: log = stack[ -2 ].log else: # FIXME: # attaching to head log is not always the right solution # if a sub-process has parallel sub-nodes, then a new thread # and a new stack will be created, but the logs will not be # appended to the correct parent log = self._processStackHead.log if log is not None: #print "Create subLog for process ", process.name newStackTop.log = log.subLog() process._log = newStackTop.log content = '
' + _t_(process.name) + '
' + _t_('Process identifier') + '
' + process._id content += '' + _t_('Execution platform') +'
' if(hasattr(process, 'executionWorkflow')): content += 'Soma-Workflow' else : content += 'BrainVISA' content += '' + _t_('Parameters') +'
' for n in process.signature.keys(): content += '' + n + ' = ' + htmlEscape( str( getattr( process, n, None ) ) ) + '' content += '