[core] Apply corrections to handle statuses correctly with pre/post process

Author: Alxiice

meshroom/core/desc/node.py

@@ -61,19 +61,6 @@ class MrNodeType(enum.Enum):
     BACKDROP = enum.auto()
 
 
-def add_method_flag(methodName: str):
-    """ Decorator used to add a flag to a method. A flag is simply a variable set to True.
-    It will be used in this context to track whether a method is overloaded or not.
-
-    Args:
-        methodName (str): Name of the annotation to set
-    """
-    def wrapper(method):
-        method.__annotations__[methodName] = True
-        return method
-    return wrapper
-
-
 class InternalAttributesFactory:
     BASIC = [
         StringParam(
@@ -285,7 +272,6 @@ def postUpdate(cls, node):
         """
         pass
 
-    @add_method_flag("disabled_preprocess")
     def preprocess(self, node):
         """ Gets invoked just before the processChunk method for the node.
 
@@ -297,9 +283,8 @@ def preprocess(self, node):
     @property
     def _hasPreprocess(self):
         """ Returns True if the class has a preprocess """
-        return not self.preprocess.__annotations__.get("disabled_preprocess", False)
+        return type(self).preprocess is not BaseNode.preprocess
 
-    @add_method_flag("disabled_postprocess")
     def postprocess(self, node):
         """ Gets invoked after the processChunk method for the node.
 
@@ -311,7 +296,7 @@ def postprocess(self, node):
     @property
     def _hasPostprocess(self):
         """ Returns True if the class has a postprocess """
-        return not self.postprocess.__annotations__.get("disabled_postprocess", False)
+        return type(self).postprocess is not BaseNode.postprocess
 
     def process(self, node):
         raise NotImplementedError(f'No process implementation on node: "{node.name}"')

meshroom/core/graph.py

@@ -1775,31 +1775,24 @@ def executeGraph(graph, toNodes=None, forceCompute=False, forceStatus=False):
         node.initStatusOnCompute(forceCompute)
 
     for n, node in enumerate(nodes):
-        print(f"(executeGraph) {n} {node}")
         try:
             # If the node is in compatibility mode, it cannot be computed
             if node.isCompatibilityNode:
                 logging.warning(f"{node.name} is in Compatibility Mode and cannot be computed: {node.issueDetails}.")
                 continue
 
-            print("A")
             node.preprocess(forceCompute)
             if not node._chunksCreated:
-                print("B create chunks")
                 node.createChunks()
             multiChunks = len(node.chunks) > 1
-            print("C")
             for c, chunk in enumerate(node.chunks):
-                print("D", c, chunk)
                 if multiChunks:
                     print('\n[{node}/{nbNodes}]({chunk}/{nbChunks}) {nodeName}'.format(
                         node=n+1, nbNodes=len(nodes),
                         chunk=c+1, nbChunks=len(node.chunks), nodeName=node.nodeType))
                 else:
                     print(f'\n[{n + 1}/{len(nodes)}] {node.nodeType}')
-                print("E")
                 chunk.process(forceCompute)
-            print("F")
             node.postprocess(forceCompute)
         except Exception as exc:
             logging.error(f"Error on node computation: {exc}")

meshroom/core/node.py

@@ -16,7 +16,7 @@
 import uuid
 from collections import namedtuple, OrderedDict
 from enum import Enum, IntEnum, auto
-from typing import Callable, Optional, List, Dict
+from typing import Callable, Optional, List, Dict, Union
 
 import meshroom
 from meshroom.common import Signal, Variant, Property, BaseObject, Slot, ListModel, DictModel
@@ -481,7 +481,7 @@ def __init__(self, node, range, parent=None):
         self.node.internalFolderChanged.connect(self.nodeFolderChanged)
 
     def __repr__(self):
-        return f"<NodeChunk {hex(id(self))}>"
+        return f"<NodeChunk {self.name} ({self.getStatusName()}) {hex(id(self))}>"
 
     @property
     def index(self):
@@ -1323,11 +1323,11 @@ def ram(self):
         return self.nodeDesc.resolvedRam(self)
 
     def hasStatus(self, status: Status):
+        if self.isInputNode:
+            return status == Status.INPUT
         if not self._chunks or not self._chunksCreated:
-            if self.isInputNode:
-                return status == Status.INPUT
             return status == Status.NONE
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             if chunk.status.status != status:
                 return False
         return True
@@ -1368,19 +1368,19 @@ def clearData(self):
     @Slot(result=str)
     def getStartDateTime(self):
         """ Return the date (str) of the first running chunk """
-        dateTime = [chunk._status.startDateTime for chunk in self._chunks if chunk._status.status
+        dateTime = [chunk._status.startDateTime for chunk in self._allChunks if chunk._status.status
                     not in (Status.NONE, Status.SUBMITTED) and chunk._status.startDateTime != ""]
         return min(dateTime) if len(dateTime) != 0 else ""
 
     def isAlreadySubmitted(self):
         if self._chunksCreated:
-            return any(c.isAlreadySubmitted() for c in self._chunks)
+            return any(c.isAlreadySubmitted() for c in self._allChunks)
         else:
             return self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING)
 
     def isAlreadySubmittedOrFinished(self):
         if self._chunksCreated:
-            return all(c.isAlreadySubmittedOrFinished() for c in self._chunks)
+            return all(c.isAlreadySubmittedOrFinished() for c in self._allChunks)
         else:
             return self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING, Status.SUCCESS)
 
@@ -1394,30 +1394,30 @@ def isSubmittedOrRunning(self):
             return False
         if not self.isAlreadySubmittedOrFinished():
             return False
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             if chunk.isRunning():
                 return True
         return False
 
     @Slot(result=bool)
     def isRunning(self):
         """ Return True if at least one chunk of this Node is running, False otherwise. """
-        return any(chunk.isRunning() for chunk in self._chunks)
+        return any(chunk.isRunning() for chunk in self._allChunks)
 
     @Slot(result=bool)
     def isFinishedOrRunning(self):
         """
         Return True if all chunks of this Node is either finished or running, False
         otherwise.
         """
-        if not self._chunks:
+        if not self._allChunks:
             return False
-        return all(chunk.isFinishedOrRunning() for chunk in self._chunks)
+        return all(chunk.isFinishedOrRunning() for chunk in self._allChunks)
 
     @Slot(result=bool)
     def isPartiallyFinished(self):
         """ Return True is at least one chunk of this Node is finished, False otherwise. """
-        return any(chunk.isFinished() for chunk in self._chunks)
+        return any(chunk.isFinished() for chunk in self._allChunks)
 
     def isExtern(self):
         """
@@ -1436,7 +1436,7 @@ def isExtern(self):
             elif self._nodeStatus.execMode == ExecMode.LOCAL and self._nodeStatus.status in (Status.SUBMITTED, Status.RUNNING):
                 return meshroom.core.sessionUid != self._nodeStatus.submitterSessionUid
             return False
-        return any(chunk.isExtern() for chunk in self._chunks)
+        return any(chunk.isExtern() for chunk in self._allChunks)
 
     @Slot()
     def clearSubmittedChunks(self):
@@ -1448,28 +1448,30 @@ def clearSubmittedChunks(self):
             This must be used with caution. This could lead to inconsistent node status
             if the graph is still being computed.
         """
-        if self._chunksCreated:
-            for chunk in self._chunks:
-                if chunk.isAlreadySubmitted():
-                    chunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
-        else:
-            if self.isAlreadySubmitted():
-                self.upgradeStatusTo(Status.NONE, ExecMode.NONE)
+        chunks: List[Union[BaseNode, NodeChunk]] = self._allChunks
+        if not self._chunksCreated:
+            chunks.append(self)
+        for chunk in chunks:
+            if chunk.isAlreadySubmitted():
+                chunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
         self.globalStatusChanged.emit()
 
     def clearLocallySubmittedChunks(self):
         """ Reset all locally submitted chunks to Status.NONE. """
-        if self._chunksCreated:
-            for chunk in self._chunks:
-                if chunk.isAlreadySubmitted() and not chunk.isExtern():
-                    chunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
-        else:
-            if self.isAlreadySubmitted() and not self.isExtern():
-                self.upgradeStatusTo(Status.NONE, ExecMode.NONE)
+        chunks: List[Union[BaseNode, NodeChunk]] = self._allChunks
+        if not self._chunksCreated:
+            chunks.append(self)
+        for chunk in chunks:
+            if chunk.isAlreadySubmitted() and not chunk.isExtern():
+                chunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
         self.globalStatusChanged.emit()
 
     def upgradeStatusTo(self, newStatus, execMode=None):
         """ Upgrade node to the given status and save it on disk. """
+        if self.nodeDesc._hasPreprocess:
+            self._preprocessChunk.upgradeStatusTo(newStatus)
+        if self.nodeDesc._hasPostprocess:
+            self._postprocessChunk.upgradeStatusTo(newStatus)
         if self._chunksCreated:
             for chunk in self._chunks:
                 chunk.upgradeStatusTo(newStatus)
@@ -1486,7 +1488,7 @@ def upgradeStatusTo(self, newStatus, execMode=None):
         self.globalStatusChanged.emit()
 
     def updateStatisticsFromCache(self):
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             chunk.updateStatisticsFromCache()
 
     def _resetChunks(self):
@@ -1672,6 +1674,10 @@ def updateStatusFromCache(self):
                 logging.warning(f"Could not create chunks from cache: {e}")
                 return
         s = self.globalStatus
+        if self.nodeDesc._hasPreprocess:
+            self._preprocessChunk.updateStatusFromCache()
+        if self.nodeDesc._hasPostprocess:
+            self._postprocessChunk.updateStatusFromCache()
         if self._chunksCreated:
             for chunk in self._chunks:
                 chunk.updateStatusFromCache()
@@ -1717,7 +1723,7 @@ def initStatusOnSubmit(self, forceCompute=False):
             #    `updateStatusFromCache()` will NOT recreate chunks from stale cache,
             #    allowing `node.createChunks()` to evaluate fresh chunk parameters.
             self._nodeStatus.resetChunkInfo()
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             if forceCompute or chunk._status.status != Status.SUCCESS:
                 hasChunkToLaunch = True
                 chunk._status.setNode(self)
@@ -1742,7 +1748,7 @@ def initStatusOnCompute(self, forceCompute=False):
             # Same rationale as initStatusOnSubmit: clear stale chunk info
             # so that the nodeStatus file does not contain outdated chunk setup.
             self._nodeStatus.resetChunkInfo()
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             if forceCompute or (chunk._status.status not in (Status.RUNNING, Status.SUCCESS)):
                 hasChunkToLaunch = True
                 chunk._status.setNode(self)
@@ -1884,6 +1890,10 @@ def endSequence(self):
 
     def stopComputation(self):
         """ Stop the computation of this node. """
+        if self.nodeDesc._hasPreprocess:
+            self._preprocessChunk.stopProcess()
+        if self.nodeDesc._hasPostprocess:
+            self._postprocessChunk.stopProcess()
         if self._chunks:
             for chunk in self._chunks.values():
                 chunk.stopProcess()
@@ -1906,12 +1916,12 @@ def getGlobalStatus(self):
         if not self._chunksCreated:
             # Get status from nodeStatus
             return self._nodeStatus.status
-        if not self._chunks:
+        if not self._allChunks:
             return Status.NONE
-        if len(self._chunks) == 1:
-            return self._chunks[0]._status.status
-
-        chunksStatus = [chunk._status.status for chunk in self._chunks]
+        if len(self._allChunks) == 1:
+            return self._allChunks[0]._status.status
+        
+        chunksStatus = [chunk._status.status for chunk in self._allChunks]
 
         anyOf = (Status.ERROR, Status.STOPPED, Status.KILLED,
                  Status.RUNNING, Status.SUBMITTED)
@@ -1928,11 +1938,11 @@ def getGlobalStatus(self):
 
     @Slot(result=ChunkStatusData)
     def getFusedStatus(self):
-        if not self._chunks:
+        if not self._allChunks:
             return ChunkStatusData()
         fusedStatus = ChunkStatusData()
-        fusedStatus.fromDict(self._chunks[0]._status.toDict())
-        for chunk in self._chunks[1:]:
+        fusedStatus.fromDict(self._allChunks[0]._status.toDict())
+        for chunk in self._allChunks[1:]:
             fusedStatus.merge(chunk._status)
         fusedStatus.status = self.getGlobalStatus()
         return fusedStatus
@@ -1958,8 +1968,8 @@ def _isBackdropNode(self) -> bool:
     def globalExecMode(self):
         if not self._chunksCreated:
             return self._nodeStatus.execMode.name
-        if len(self._chunks):
-            return self._chunks.at(0).getExecModeName()
+        if len(self._allChunks):
+            return self._allChunks[0].getExecModeName()
         else:
             return ExecMode.NONE
 
@@ -1974,7 +1984,17 @@ def _getJobName(self):
 
     def getChunks(self) -> list[NodeChunk]:
         return self._chunks
-    
+
+    @property
+    def _allChunks(self) -> list[NodeChunk]:
+        chunks = []
+        if self.nodeDesc._hasPreprocess:
+            chunks.append(self._preprocessChunk)
+        chunks.extend([c for c in self._chunks])
+        if self.nodeDesc._hasPostprocess:
+            chunks.append(self._postprocessChunk)
+        return chunks
+
     def getAllChunks(self):
         allChunks = []
         if self.nodeDesc._hasPreprocess:
@@ -2097,19 +2117,19 @@ def updateDuplicates(self, nodesPerUid):
 
     def initFromThisSession(self) -> bool:
         """ Check if the node was submitted from the current session """
-        if not self._chunksCreated or not self._chunks:
+        if not self._chunksCreated or not self._allChunks:
             return meshroom.core.sessionUid == self._nodeStatus.submitterSessionUid
-        for chunk in self._chunks:
+        for chunk in self._allChunks:
             # Technically the check on chunk._status.computeSessionUid is useless
             if meshroom.core.sessionUid not in (chunk._status.computeSessionUid, self._nodeStatus.submitterSessionUid):
                 return False
         return True
 
     def isMainNode(self) -> bool:
         """ In case of a node with duplicates, we check that the node is the one driving the computation. """
-        if len(self._chunks) == 0:
+        if len(self._allChunks) == 0:
             return True
-        firstChunk = self._chunks.at(0)
+        firstChunk = self._allChunks[0]
         if not firstChunk.statusNodeName:
             # If nothing is declared, anyone could become the main (if there are duplicates).
             return True