Pre/Post tasks : Fix typos & style + remove dead code

Author: Alxiice

bin/meshroom_compute

@@ -50,7 +50,8 @@ parser.add_argument('-v', '--verbose',
                     default=os.environ.get('MESHROOM_VERBOSE', 'info'),
                     choices=['fatal', 'error', 'warning', 'info', 'debug', 'trace'])
 
-parser.add_argument('-i', '--iteration', type=int, default=ChunkIndex.NONE, help='')
+parser.add_argument('-i', '--iteration', type=int, default=ChunkIndex.NONE, 
+                    help='Define specific chunk index to compute')
 parser.add_argument('--preprocess', help='Execute preprocess chunk', action='store_true')
 parser.add_argument('--postprocess', help='Execute postprocess chunk', action='store_true')
 
@@ -151,8 +152,7 @@ if args.node:
         for chunk in chunks:
             if chunk.status.status in submittedStatuses:
                 # Particular case for the local isolated, the node status is set to RUNNING by the submitter directly.
-                # We ensure that no other instance has start
-                # ed to compute, by checking that the computeSessionUid is empty.
+                # We ensure that no other instance has started to compute, by checking that the computeSessionUid is empty.
                 if chunk.node.getMrNodeType() == meshroom.core.MrNodeType.NODE and \
                     not chunk.status.computeSessionUid and node._nodeStatus.submitterSessionUid:
                     continue

meshroom/core/desc/node.py

@@ -281,7 +281,7 @@ def preprocess(self, node):
         pass
 
     @property
-    def _hasPreprocess(self):
+    def hasPreprocess(self):
         """ Returns True if the class has a preprocess """
         return type(self).preprocess is not BaseNode.preprocess
 
@@ -294,7 +294,7 @@ def postprocess(self, node):
         pass
 
     @property
-    def _hasPostprocess(self):
+    def hasPostprocess(self):
         """ Returns True if the class has a postprocess """
         return type(self).postprocess is not BaseNode.postprocess
 

meshroom/core/node.py

@@ -845,9 +845,9 @@ def __init__(self, nodeType: str, position: Position = None, parent: BaseObject
         self.dirty: bool = True  # whether this node's outputs must be re-evaluated on next Graph update
         self._chunks: list[NodeChunk] = ListModel(parent=self)
         self._preprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.PREPROCESS)) if \
-            self.nodeDesc and self.nodeDesc._hasPreprocess else None
+            self.nodeDesc and self.nodeDesc.hasPreprocess else None
         self._postprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.POSTPROCESS)) if \
-            self.nodeDesc and self.nodeDesc._hasPostprocess else None
+            self.nodeDesc and self.nodeDesc.hasPostprocess else None
         self._chunksCreated = False  # Only initialize chunks on compute
         self._chunkPlaceholder: list[NodeChunk] = ListModel(parent=self)  # Placeholder chunk for nodes with dynamic ones
         self._uid: str = uid
@@ -1468,9 +1468,9 @@ def clearLocallySubmittedChunks(self):
 
     def upgradeStatusTo(self, newStatus, execMode=None):
         """ Upgrade node to the given status and save it on disk. """
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             self._preprocessChunk.upgradeStatusTo(newStatus)
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             self._postprocessChunk.upgradeStatusTo(newStatus)
         if self._chunksCreated:
             for chunk in self._chunks:
@@ -1674,9 +1674,9 @@ def updateStatusFromCache(self):
                 logging.warning(f"Could not create chunks from cache: {e}")
                 return
         s = self.globalStatus
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             self._preprocessChunk.updateStatusFromCache()
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             self._postprocessChunk.updateStatusFromCache()
         if self._chunksCreated:
             for chunk in self._chunks:
@@ -1780,7 +1780,7 @@ def processIteration(self, iteration):
 
     def preprocess(self, forceCompute=False, inCurrentEnv=False):
         """ Prepare the node processing """
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             self.prepareLogger(ChunkIndex.PREPROCESS)
             self._preprocessChunk.process(forceCompute, inCurrentEnv)
             self.restoreLogger()
@@ -1794,7 +1794,7 @@ def postprocess(self, forceCompute=False, inCurrentEnv=False):
         Invoke the post process on Client Node to execute after the processing on the
         node is completed
         """
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             self.prepareLogger(ChunkIndex.POSTPROCESS)
             self._postprocessChunk.process(forceCompute, inCurrentEnv)
             self.restoreLogger()
@@ -1904,9 +1904,9 @@ def endSequence(self):
 
     def stopComputation(self):
         """ Stop the computation of this node. """
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             self._preprocessChunk.stopProcess()
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             self._postprocessChunk.stopProcess()
         if self._chunks:
             for chunk in self._chunks.values():
@@ -2002,19 +2002,19 @@ def getChunks(self) -> list[NodeChunk]:
     @property
     def _allChunks(self) -> list[NodeChunk]:
         chunks = []
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             chunks.append(self._preprocessChunk)
         chunks.extend([c for c in self._chunks])
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             chunks.append(self._postprocessChunk)
         return chunks
 
     def getAllChunks(self):
         allChunks = []
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             allChunks.append({"chunkIndex": ChunkIndex.PREPROCESS, "chunk": self._preprocessChunk, "name": "Preprocess"})
         allChunks.extend([{"chunkIndex": i, "chunk": c, "name": str(i)} for i, c in enumerate(self._chunks)])
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             allChunks.append({"chunkIndex": ChunkIndex.POSTPROCESS, "chunk": self._postprocessChunk, "name": "Postprocess"})
         return allChunks
 
@@ -2277,6 +2277,8 @@ def _hasDisplayableShape(self):
     chunksCreated = Property(bool, lambda self: self._chunksCreated, notify=chunksCreatedChanged)
     chunksChanged = Signal()
     chunks = Property(Variant, getChunks, notify=chunksChanged)
+    preprocessChunk = Property(Variant, lambda self: self._preprocessChunk, notify=chunksChanged)
+    postprocessChunk = Property(Variant, lambda self: self._postprocessChunk, notify=chunksChanged)
     allChunks = Property(Variant, getAllChunks, notify=chunksChanged)
     chunkPlaceholder = Property(Variant, lambda self: self._chunkPlaceholder, notify=chunksChanged)
     nbParallelizationBlocks = Property(int, lambda self: len(self._chunks) if self._chunksCreated else 0, notify=chunksChanged)
@@ -2470,12 +2472,12 @@ def _resetChunks(self):
             self.setSize(0)
             self._chunkPlaceholder.setObjectList([NodeChunk(self, desc.computation.Range())])
         # Pre/post process
-        if self.nodeDesc._hasPreprocess:
+        if self.nodeDesc.hasPreprocess:
             self._preprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.PREPROCESS))
             self._preprocessChunk.statusChanged.connect(self.globalStatusChanged)
         else:
             self._preprocessChunk = None
-        if self.nodeDesc._hasPostprocess:
+        if self.nodeDesc.hasPostprocess:
             self._postprocessChunk = NodeChunk(self, desc.Range(ChunkIndex.POSTPROCESS))
             self._postprocessChunk.statusChanged.connect(self.globalStatusChanged)
         else:

meshroom/core/submitter.py

@@ -132,7 +132,7 @@ def chunksIterations(self) -> list[int]:
         """
         if self.isExpanding:
             return None
-        if not self.isExpanding and self.node.isParallelized:
+        if self.node.isParallelized:
             _, _, nbBlocks = self.node.nodeDesc.parallelization.getSizes(self.node)
             iterationsToIgnore = []
             for c in self.node._chunks:
@@ -144,11 +144,11 @@ def chunksIterations(self) -> list[int]:
 
     @property
     def hasPreprocess(self) -> bool:
-        return self.node.nodeDesc._hasPreprocess
+        return self.node.nodeDesc.hasPreprocess
 
     @property
     def hasPostprocess(self) -> bool:
-        return self.node.nodeDesc._hasPostprocess
+        return self.node.nodeDesc.hasPostprocess
 
     def __repr__(self):
         depsNames = "|".join([t.node.name for t in self.dependencies])
@@ -181,7 +181,7 @@ def __init__(self, nodes, edges):
 
     def _orderNodes(self, nodes, edges):
         """
-        Take all the nodes and connections and orger them by processing step
+        Take all the nodes and connections and order them by processing step
         0 is the root nodes (can be executed last)
         Then 1 is the level with the direct dependencies for the root nodes, and etc...
         """

meshroom/core/taskManager.py

@@ -75,27 +75,6 @@ def onChunksCreated(createdNode):
             self._manager.chunksCreated.disconnect(onChunksCreated)
             timer.stop()
 
-    def execChunk(self, chunk, node, function) -> bool:
-        """ Handle the chunk process & fail
-
-        Args:
-            chunk (chunk to process): _description_
-            node (_type_): _description_
-            function (_type_): _description_
-
-        Returns:
-            bool: _description_
-        """
-        try:
-            function(self.forceCompute)
-        except Exception as exc:
-            if chunk.isStopped():
-                return True
-            else:
-                logging.error(f"Error on node computation: {exc}")
-                self.clearNodes(node)
-        return False
-    
     def clearNodes(self, node):
         nodesToRemove, _ = self._manager._graph.dfsOnDiscover(startNodes=[node], reverse=True)
         # remove following nodes from the task queue
@@ -152,7 +131,7 @@ def run(self):
                 for chunk in node._chunks:
                     if chunk.isAlreadySubmitted():
                         chunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
-                if node.nodeDesc._hasPostprocess:
+                if node.nodeDesc.hasPostprocess:
                     node._postprocessChunk.upgradeStatusTo(Status.NONE, ExecMode.NONE)
                 break
             # Process

tests/test_compute.py

@@ -191,15 +191,14 @@ def test_prepostprocess(self, tmp_path):
         root = chunkLog.parent
         preprocessLog = root / "preprocess.log"
         postprocessLog = root / "postprocess.log"
-        def check_file(path, suffix = ""):
+        def check_file(path, suffix=""):
             with open(path, "r") as f:
                 content = f.read()
-                suffix = ""
                 reg = re.compile(self.logPrefix + "TestNodeD_1" + suffix)
                 assert len(reg.findall(content)) == 1
-        check_file(preprocessLog, " (preprocess)")
+        check_file(preprocessLog, " \(preprocess\)")
         check_file(chunkLog, "")
-        check_file(postprocessLog, " postprocess")
+        check_file(postprocessLog, " \(postprocess\)")
 
 
 class TestLockUpdates:
@@ -568,8 +567,8 @@ def test_status(self, graphSavedOnDisk):
 
         # Check node
         assert len(node.chunks) == 1
-        assert node.nodeDesc._hasPreprocess
-        assert node.nodeDesc._hasPostprocess
+        assert node.nodeDesc.hasPreprocess
+        assert node.nodeDesc.hasPostprocess
 
         # Check status before
         assert node.globalStatus == Status.NONE.name

tests/test_submit.py

@@ -171,7 +171,6 @@ def test_orderTasks(self):
         nodeB = self.addNewNode(graph, "PluginSubmitter"+"B", nodeParams={"inputs": [nodeA.output]})
         nodeC = self.addNewNode(graph, "PluginSubmitter"+"C", nodeParams={"inputs": [nodeB.output]})
         # Order tasks
-        submitter = get_submitter()
         nodes, edges = graph.dfsOnFinish(startNodes=[nodeC])
         orderedTasks = OrderedTasks(nodes, edges)
         # === Test result ===