Add chunk notion for parallelization and implement specific updateInternals in CameraInit node

* Add chunk notion for parallelization
* Allows Node desc to implement custom updateInternals
* CameraInit node implement a specific updateInternals to update the
input image list
* FeatureExtraction, FeatureMatching, DepthMap, DepthMapFilter:
implement parallelization

meshroom/core/graph.py

@@ -386,8 +386,9 @@ class Status(Enum):
     SUBMITTED_LOCAL = 3
     RUNNING = 4
     ERROR = 5
-    KILLED = 6
-    SUCCESS = 7
+    STOPPED = 6
+    KILLED = 7
+    SUCCESS = 8
 
 
 class StatusData:
@@ -429,6 +430,143 @@ def clearProcessesStatus():
         v.upgradeStatusTo(Status.KILLED)
 
 
+class NodeChunk(BaseObject):
+    def __init__(self, node, range):
+        super(NodeChunk, self).__init__(node)
+        self.node = node
+        self.range = range
+        self.status = StatusData(node.name, node.nodeType)
+        self.statistics = stats.Statistics()
+        self._subprocess = None
+
+    @property
+    def index(self):
+        return self.range.iteration
+
+    @property
+    def name(self):
+        if self.range.blockSize:
+            return "{}({})".format(self.node.name, self.index)
+        else:
+            return self.node.name
+
+    @property
+    def statusName(self):
+        return self.status.name
+
+    def updateStatusFromCache(self):
+        """
+        Update node status based on status file content/existence.
+        """
+        statusFile = self.statusFile()
+        oldStatus = self.status.status
+        # No status file => reset status to Status.None
+        if not os.path.exists(statusFile):
+            self.status.reset()
+        else:
+            with open(statusFile, 'r') as jsonFile:
+                statusData = json.load(jsonFile)
+            self.status.fromDict(statusData)
+        if oldStatus != self.status.status:
+            self.statusChanged.emit()
+
+    def statusFile(self):
+        if self.range.blockSize == 0:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, 'status')
+        else:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, str(self.index) + '.status')
+
+    def statisticsFile(self):
+        if self.range.blockSize == 0:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, 'statistics')
+        else:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, str(self.index) + '.statistics')
+
+    def logFile(self):
+        if self.range.blockSize == 0:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, 'log')
+        else:
+            return os.path.join(self.node.graph.cacheDir, self.node.internalFolder, str(self.index) + '.log')
+
+    def saveStatusFile(self):
+        """
+        Write node status on disk.
+        """
+        data = self.status.toDict()
+        statusFilepath = self.statusFile()
+        folder = os.path.dirname(statusFilepath)
+        if not os.path.exists(folder):
+            os.makedirs(folder)
+        statusFilepathWriting = statusFilepath + '.writing.' + str(uuid.uuid4())
+        with open(statusFilepathWriting, 'w') as jsonFile:
+            json.dump(data, jsonFile, indent=4)
+        shutil.move(statusFilepathWriting, statusFilepath)
+
+    def upgradeStatusTo(self, newStatus):
+        if newStatus.value <= self.status.status.value:
+            print('WARNING: downgrade status on node "{}" from {} to {}'.format(self.name, self.status.status,
+                                                                                newStatus))
+        self.status.status = newStatus
+        self.statusChanged.emit()
+        self.saveStatusFile()
+
+    def updateStatisticsFromCache(self):
+        """
+        """
+        oldTimes = self.statistics.times
+        statisticsFile = self.statisticsFile()
+        if not os.path.exists(statisticsFile):
+            return
+        with open(statisticsFile, 'r') as jsonFile:
+            statisticsData = json.load(jsonFile)
+        self.statistics.fromDict(statisticsData)
+        if oldTimes != self.statistics.times:
+            self.statisticsChanged.emit()
+
+    def saveStatistics(self):
+        data = self.statistics.toDict()
+        statisticsFilepath = self.statisticsFile()
+        folder = os.path.dirname(statisticsFilepath)
+        if not os.path.exists(folder):
+            os.makedirs(folder)
+        statisticsFilepathWriting = statisticsFilepath + '.writing.' + str(uuid.uuid4())
+        with open(statisticsFilepathWriting, 'w') as jsonFile:
+            json.dump(data, jsonFile, indent=4)
+        shutil.move(statisticsFilepathWriting, statisticsFilepath)
+
+    def isAlreadySubmitted(self):
+        return self.status.status in (Status.SUBMITTED_EXTERN, Status.SUBMITTED_LOCAL, Status.RUNNING)
+
+    def process(self):
+        global runningProcesses
+        runningProcesses[self.name] = self
+        self.upgradeStatusTo(Status.RUNNING)
+        self.statThread = stats.StatisticsThread(self)
+        self.statThread.start()
+        startTime = time.time()
+        try:
+            self.node.nodeDesc.processChunk(self)
+        except Exception as e:
+            self.upgradeStatusTo(Status.ERROR)
+            raise
+        except (KeyboardInterrupt, SystemError, GeneratorExit) as e:
+            self.upgradeStatusTo(Status.STOPPED)
+            raise
+        finally:
+            elapsedTime = time.time() - startTime
+            print(' - elapsed time:', elapsedTime)
+            # ask and wait for the stats thread to stop
+            self.statThread.stopRequest()
+            self.statThread.join()
+            del runningProcesses[self.name]
+
+        self.upgradeStatusTo(Status.SUCCESS)
+
+    statusChanged = Signal()
+    statusName = Property(str, statusName.fget, notify=statusChanged)
+    statisticsChanged = Signal()
+
+
 class Node(BaseObject):
     """
     """
@@ -439,11 +577,13 @@ class Node(BaseObject):
 
     def __init__(self, nodeDesc, parent=None, **kwargs):
         super(Node, self).__init__(parent)
-        self._name = None  # type: str
-        self.graph = None  # type: Graph
         self.nodeDesc = meshroom.core.nodesDesc[nodeDesc]()
         self.packageName = self.nodeDesc.packageName
         self.packageVersion = self.nodeDesc.packageVersion
+
+        self._name = None  # type: str
+        self.graph = None  # type: Graph
+        self._chunks = []
         self._cmdVars = {}
         self._attributes = DictModel(keyAttrName='name', parent=self)
         self.attributesPerUid = defaultdict(set)
@@ -451,9 +591,6 @@ class Node(BaseObject):
         for k, v in kwargs.items():
             self.attribute(k).value = v
 
-        self.status = StatusData(self.name, self.nodeType)
-        self.statistics = stats.Statistics()
-        self._subprocess = None
 
     def __getattr__(self, k):
         try:
@@ -538,15 +675,11 @@ class Node(BaseObject):
             'attributes': {k: v for k, v in attributes.items() if v is not None},  # filter empty values
         }
 
-    def updateInternals(self):
-        self._cmdVars = {
-            'cache': self.graph.cacheDir,
-            'nodeType': self.nodeType,
-        }
+    def _buildCmdVars(self, cmdVars):
         for uidIndex, associatedAttributes in self.attributesPerUid.items():
             assAttr = [(a.getName(), a.uid(uidIndex)) for a in associatedAttributes]
             assAttr.sort()
-            self._cmdVars['uid{}'.format(uidIndex)] = hash(tuple([b for a, b in assAttr]))
+            cmdVars['uid{}'.format(uidIndex)] = hash(tuple([b for a, b in assAttr]))
 
         # Evaluate input params
         for name, attr in self._attributes.objects.items():
@@ -557,15 +690,15 @@ class Node(BaseObject):
                 assert(isinstance(v, collections.Sequence) and not isinstance(v, basestring))
                 v = attr.attributeDesc.joinChar.join(v)
 
-            self._cmdVars[name] = '--{name} {value}'.format(name=name, value=v)
-            self._cmdVars[name + 'Value'] = str(v)
+            cmdVars[name] = '--{name} {value}'.format(name=name, value=v)
+            cmdVars[name + 'Value'] = str(v)
 
             if v is not None and v is not '':
-                self._cmdVars[attr.attributeDesc.group] = self._cmdVars.get(attr.attributeDesc.group, '') + \
-                                                          ' ' + self._cmdVars[name]
+                cmdVars[attr.attributeDesc.group] = cmdVars.get(attr.attributeDesc.group, '') + \
+                                                          ' ' + cmdVars[name]
 
         # For updating output attributes invalidation values
-        cmdVarsNoCache = self._cmdVars.copy()
+        cmdVarsNoCache = cmdVars.copy()
         cmdVarsNoCache['cache'] = ''
 
         # Evaluate output params
@@ -573,17 +706,81 @@ class Node(BaseObject):
             if attr.isInput:
                 continue  # skip inputs
             attr.value = attr.attributeDesc.value.format(
-                **self._cmdVars)
+                **cmdVars)
             attr._invalidationValue = attr.attributeDesc.value.format(
                 **cmdVarsNoCache)
             v = attr.value
 
-            self._cmdVars[name] = '--{name} {value}'.format(name=name, value=v)
-            self._cmdVars[name + 'Value'] = str(v)
+            cmdVars[name] = '--{name} {value}'.format(name=name, value=v)
+            cmdVars[name + 'Value'] = str(v)
 
             if v is not None and v is not '':
-                self._cmdVars[attr.attributeDesc.group] = self._cmdVars.get(attr.attributeDesc.group, '') + \
-                                                          ' ' + self._cmdVars[name]
+                cmdVars[attr.attributeDesc.group] = cmdVars.get(attr.attributeDesc.group, '') + \
+                                                          ' ' + cmdVars[name]
+
+    @property
+    def isParallelized(self):
+        return bool(self.nodeDesc.parallelization)
+
+    @property
+    def nbParallelizationBlocks(self):
+        return len(self.chunks)
+
+    def hasStatus(self, status):
+        if not self.chunks:
+            return False
+        for chunk in self.chunks:
+            if chunk.status.status != status:
+                return False
+        return True
+
+    def isAlreadySubmitted(self):
+        for chunk in self.chunks:
+            if chunk.isAlreadySubmitted():
+                return True
+        return False
+
+    def alreadySubmittedChunks(self):
+        submittedChunks = []
+        for chunk in self.chunks:
+            if chunk.isAlreadySubmitted():
+                submittedChunks.append(chunk)
+        return submittedChunks
+
+    def upgradeStatusTo(self, newStatus):
+        """
+        Upgrade node to the given status and save it on disk.
+        """
+        for chunk in self.chunks:
+            chunk.upgradeStatusTo(newStatus)
+
+    def updateStatisticsFromCache(self):
+        for chunk in self.chunks:
+            chunk.updateStatisticsFromCache()
+
+    def updateInternals(self):
+        if self.isParallelized:
+            ranges = self.nodeDesc.parallelization.getRanges(self)
+            if len(ranges) != len(self.chunks):
+                self._chunks = [NodeChunk(self, range) for range in ranges]
+                self.chunksChanged.emit()
+            else:
+                for chunk, range in zip(self.chunks, ranges):
+                    chunk.range = range
+        else:
+            if len(self._chunks) != 1:
+                self._chunks = [NodeChunk(self, desc.Range())]
+                self.chunksChanged.emit()
+            else:
+                self._chunks[0].range = desc.Range()
+
+        self._cmdVars = {
+            'cache': self.graph.cacheDir,
+            'nodeType': self.nodeType,
+        }
+        self._buildCmdVars(self._cmdVars)
+
+        self.nodeDesc.updateInternals(self)
 
         self.internalFolderChanged.emit()
 
@@ -591,79 +788,12 @@ class Node(BaseObject):
     def internalFolder(self):
         return self.nodeDesc.internalFolder.format(**self._cmdVars)
 
-    def statusFile(self):
-        return os.path.join(self.graph.cacheDir, self.internalFolder, 'status')
-
-    def statisticsFile(self):
-        return os.path.join(self.graph.cacheDir, self.internalFolder, 'statistics')
-
-    def logFile(self):
-        return os.path.join(self.graph.cacheDir, self.internalFolder, 'log')
-
     def updateStatusFromCache(self):
         """
         Update node status based on status file content/existence.
         """
-        statusFile = self.statusFile()
-        oldStatus = self.status.status
-        # No status file => reset status to Status.None
-        if not os.path.exists(statusFile):
-            self.status.reset()
-        else:
-            with open(statusFile, 'r') as jsonFile:
-                statusData = json.load(jsonFile)
-            self.status.fromDict(statusData)
-        if oldStatus != self.status.status:
-            self.statusChanged.emit()
-
-    def saveStatusFile(self):
-        """
-        Write node status on disk.
-        """
-        data = self.status.toDict()
-        statusFilepath = self.statusFile()
-        folder = os.path.dirname(statusFilepath)
-        if not os.path.exists(folder):
-            os.makedirs(folder)
-        statusFilepathWriting = statusFilepath + '.writing.' + str(uuid.uuid4())
-        with open(statusFilepathWriting, 'w') as jsonFile:
-            json.dump(data, jsonFile, indent=4)
-        shutil.move(statusFilepathWriting, statusFilepath)
-
-    def upgradeStatusTo(self, newStatus):
-        """
-        Upgrade node to the given status and save it on disk.
-        """
-        if newStatus.value <= self.status.status.value:
-            print('WARNING: downgrade status on node "{}" from {} to {}'.format(self._name, self.status.status,
-                                                                                newStatus))
-        self.status.status = newStatus
-        self.statusChanged.emit()
-        self.saveStatusFile()
-
-    def updateStatisticsFromCache(self):
-        """
-        """
-        statisticsFile = self.statisticsFile()
-        if not os.path.exists(statisticsFile):
-            return
-        with open(statisticsFile, 'r') as jsonFile:
-            statisticsData = json.load(jsonFile)
-        self.statistics.fromDict(statisticsData)
-
-    def saveStatistics(self):
-        data = self.statistics.toDict()
-        statisticsFilepath = self.statisticsFile()
-        folder = os.path.dirname(statisticsFilepath)
-        if not os.path.exists(folder):
-            os.makedirs(folder)
-        statisticsFilepathWriting = statisticsFilepath + '.writing.' + str(uuid.uuid4())
-        with open(statisticsFilepathWriting, 'w') as jsonFile:
-            json.dump(data, jsonFile, indent=4)
-        shutil.move(statisticsFilepathWriting, statisticsFilepath)
-
-    def isAlreadySubmitted(self):
-        return self.status.status in (Status.SUBMITTED_EXTERN, Status.SUBMITTED_LOCAL, Status.RUNNING)
+        for chunk in self.chunks:
+            chunk.updateStatusFromCache()
 
     def submit(self):
         self.upgradeStatusTo(Status.SUBMITTED_EXTERN)
@@ -674,27 +804,12 @@ class Node(BaseObject):
     def stopProcess(self):
         self.nodeDesc.stop(self)
 
-    def process(self):
-        global runningProcesses
-        runningProcesses[self.name] = self
-        self.upgradeStatusTo(Status.RUNNING)
-        self.statThread = stats.StatisticsThread(self)
-        self.statThread.start()
-        startTime = time.time()
-        try:
-            self.nodeDesc.process(self)
-        except BaseException:
-            self.upgradeStatusTo(Status.ERROR)
-            raise
-        finally:
-            elapsedTime = time.time() - startTime
-            print(' - elapsed time:', elapsedTime)
-            # ask and wait for the stats thread to stop
-            self.statThread.stopRequest()
-            self.statThread.join()
-            del runningProcesses[self.name]
+    def processIteration(self, iteration):
+        self.chunks[iteration].process()
 
-        self.upgradeStatusTo(Status.SUCCESS)
+    def process(self):
+        for chunk in self.chunks:
+            chunk.process()
 
     def endSequence(self):
         pass
@@ -702,9 +817,12 @@ class Node(BaseObject):
     def getStatus(self):
         return self.status
 
+    def getChunks(self):
+        return self._chunks
+
     @property
-    def statusName(self):
-        return self.status.status.name
+    def statusNames(self):
+        return [s.status.name for s in self.status]
 
     def __repr__(self):
         return self.name
@@ -716,9 +834,8 @@ class Node(BaseObject):
     internalFolder = Property(str, internalFolder.fget, notify=internalFolderChanged)
     depthChanged = Signal()
     depth = Property(int, depth.fget, notify=depthChanged)
-    statusChanged = Signal()
-    statusName = Property(str, statusName.fget, notify=statusChanged)
-
+    chunksChanged = Signal()
+    chunks = Property(Variant, getChunks, notify=chunksChanged)
 
 WHITE = 0
 GRAY = 1
@@ -794,14 +911,16 @@ class Graph(BaseObject):
         if not isinstance(graphData, dict):
             raise RuntimeError('loadGraph error: Graph is not a dict. File: {}'.format(filepath))
 
-        self.cacheDir = os.path.join(os.path.abspath(os.path.dirname(filepath)), meshroom.core.cacheFolderName)
-        self.name = os.path.splitext(os.path.basename(filepath))[0]
-        for nodeName, nodeData in graphData.items():
-            if not isinstance(nodeData, dict):
-                raise RuntimeError('loadGraph error: Node is not a dict. File: {}'.format(filepath))
-            n = Node(nodeData['nodeType'], parent=self, **nodeData['attributes'])
-            # Add node to the graph with raw attributes values
-            self._addNode(n, nodeName)
+        with GraphModification(self):
+            # Init name and cache directory from input filepath
+            self.cacheDir = os.path.join(os.path.abspath(os.path.dirname(filepath)), meshroom.core.cacheFolderName)
+            self.name = os.path.splitext(os.path.basename(filepath))[0]
+            for nodeName, nodeData in graphData.items():
+                if not isinstance(nodeData, dict):
+                    raise RuntimeError('loadGraph error: Node is not a dict. File: {}'.format(filepath))
+                n = Node(nodeData['nodeType'], parent=self, **nodeData['attributes'])
+                # Add node to the graph with raw attributes values
+                self._addNode(n, nodeName)
 
         # Create graph edges by resolving attributes expressions
         self._applyExpr()
@@ -1036,16 +1155,22 @@ class Graph(BaseObject):
         visitor = Visitor()
 
         def finishVertex(vertex, graph):
-            if vertex.status.status in (Status.SUBMITTED_EXTERN,
-                                        Status.SUBMITTED_LOCAL):
-                print('WARNING: node "{}" is already submitted.'.format(vertex.name))
-            if vertex.status.status is Status.RUNNING:
-                print('WARNING: node "{}" is already running.'.format(vertex.name))
-            if vertex.status.status is not Status.SUCCESS:
-                nodes.append(vertex)
+            chunksToProcess = []
+            for chunk in vertex.chunks:
+                if chunk.status.status in (Status.SUBMITTED_EXTERN,
+                                            Status.SUBMITTED_LOCAL):
+                    logging.warning('Node "{}" is already submitted.'.format(vertex.name))
+                if chunk.status.status is Status.RUNNING:
+                    logging.warning('Node "{}" is already running.'.format(vertex.name))
+                if chunk.status.status is not Status.SUCCESS:
+                    chunksToProcess.append(chunk)
+            if chunksToProcess:
+                nodes.append(vertex)  # We could collect specific chunks
 
         def finishEdge(edge, graph):
-            if (edge[0].status.status is not Status.SUCCESS) and (edge[1].status.status is not Status.SUCCESS):
+            if edge[0].hasStatus(Status.SUCCESS) or edge[1].hasStatus(Status.SUCCESS):
+                return
+            else:
                 edges.append(edge)
 
         visitor.finishVertex = finishVertex
@@ -1171,13 +1296,11 @@ class Graph(BaseObject):
         """ Request graph execution to be stopped """
         self.stopExecutionRequested.emit()
 
-    def submittedNodes(self):
-        """ Return the list of submitted nodes inside this Graph """
-        return [node for node in self.nodes if node.isAlreadySubmitted()]
-
     def clearSubmittedNodes(self):
         """ Reset the status of already submitted nodes to Status.NONE """
-        [node.upgradeStatusTo(Status.NONE) for node in self.submittedNodes()]
+        for node in self.nodes:
+            for chunk in node.alreadySubmittedChunks():
+                chunk.upgradeStatusTo(Status.NONE)
 
     @property
     def nodes(self):
@@ -1222,25 +1345,26 @@ def getAlreadySubmittedNodes(nodes):
     return out
 
 
-def execute(graph, toNodes=None, force=False):
+def execute(graph, toNodes=None, forceCompute=False, forceStatus=False):
     """
     """
-    if force:
+    if forceCompute:
         nodes, edges = graph.dfsOnFinish(startNodes=toNodes)
     else:
         nodes, edges = graph.dfsToProcess(startNodes=toNodes)
         nodesInConflict = getAlreadySubmittedNodes(nodes)
 
         if nodesInConflict:
-            nodesStatus = set([node.status.status.name for node in nodesInConflict])
+            nodesStatus = set([status.status.name for node in nodesInConflict for status in node.status])
             nodesName = [node.name for node in nodesInConflict]
-            #raise RuntimeError(
-            print(
-                'WARNING: Some nodes are already submitted with status: {}\n'
-                'Nodes: {}'.format(
-                ', '.join(nodesStatus),
-                ', '.join(nodesName),
-                ))
+            msg = 'WARNING: Some nodes are already submitted with status: {}\nNodes: {}'.format(
+                  ', '.join(nodesStatus),
+                  ', '.join(nodesName)
+                  )
+            if forceStatus:
+                print(msg)
+            else:
+                raise RuntimeError(msg)
 
     print('Nodes to execute: ', str([n.name for n in nodes]))
 
@@ -1254,7 +1378,7 @@ def execute(graph, toNodes=None, force=False):
         except Exception as e:
             logging.error("Error on node computation: {}".format(e))
             graph.clearSubmittedNodes()
-            return
+            raise
 
     for node in nodes:
         node.endSequence()