wip

Author: Alxiice

meshroom/core/desc/node.py

@@ -1,3 +1,5 @@
+# desc/node.py
+
 import enum
 from inspect import getfile
 from pathlib import Path

meshroom/core/graph.py

@@ -1779,24 +1779,31 @@ 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()
             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()
         except Exception as exc:
             logging.error(f"Error on node computation: {exc}")

meshroom/core/node.py

@@ -1,4 +1,7 @@
 #!/usr/bin/env python
+
+# core/node.py
+
 import sys
 import atexit
 import copy
@@ -657,13 +660,8 @@ def isFinished(self):
         return self._status.status == Status.SUCCESS
 
     def process(self, forceCompute=False, inCurrentEnv=False):
-        if self.isPreprocess:
-            return self.node.nodeDesc.preprocess(self.node)
-        if self.isPostprocess:
-            return self.node.nodeDesc.postprocess(self.node)
-        return self._process(forceCompute, inCurrentEnv)
+        print(f"(NodeChunk) process {self} -> pre={self.isPreprocess}, post={self.isPostprocess}")
 
-    def _process(self, forceCompute=False, inCurrentEnv=False):
         if not forceCompute and self._status.status == Status.SUCCESS:
             logging.info(f"Node chunk already computed: {self.name}")
             return
@@ -672,33 +670,51 @@ def _process(self, forceCompute=False, inCurrentEnv=False):
         # This only happens once, when the node has the SUBMITTED status.
         # The sub-process will go through this method again, but the node status will
         # have been set to RUNNING.
+        print(f"(_process) A")
         if not inCurrentEnv and self.node.getMrNodeType() == MrNodeType.NODE:
+            print(f"(_process) B")
             self._processInIsolatedEnvironment()
             return
 
+        print(f"(_process) C")
         runningProcesses[self.name] = self
+        print(f"(_process) D")
         self._status.setNode(self.node)
+        print(f"(_process) E")
         self._status.initStartCompute()
+        print(f"(_process) F")
         self.upgradeStatusFile()
+        print(f"(_process) G")
         executionStatus = None
+        print(f"(_process) H")
         self.statThread = stats.StatisticsThread(self)
+        print(f"(_process) I")
         self.statThread.start()
 
+        print(f"(_process) J")
         try:
-            logging.info(f"[Process chunk] Start processing...")
-            self.node.nodeDesc.processChunk(self)
+            print(f"(_process) K")
+            if self.isPreprocess:
+                self.node.nodeDesc.preprocess(self.node)
+            elif self.isPostprocess:
+                self.node.nodeDesc.postprocess(self.node)
+            else:
+                self.node.nodeDesc.processChunk(self)
             # NOTE: this assumes saving the output attributes for each chunk
             self.node.saveOutputAttr()
             executionStatus = Status.SUCCESS
         except Exception:
+            print(f"(_process) L")
             self.updateStatusFromCache()  # check if the status has been updated by another process
             if self._status.status != Status.STOPPED:
                 executionStatus = Status.ERROR
             raise
         except (KeyboardInterrupt, SystemError, GeneratorExit):
+            print(f"(_process) M")
             executionStatus = Status.STOPPED
             raise
         finally:
+            print(f"(_process) N")
             self._status.setNode(self.node)
             self._status.initEndCompute()
             self.upgradeStatusFile()
@@ -712,6 +728,7 @@ def _process(self, forceCompute=False, inCurrentEnv=False):
             self.statistics = stats.Statistics()
             del runningProcesses[self.name]
 
+        print(f"(_process) O")
 
     def _processInIsolatedEnvironment(self):
         """

meshroom/core/taskManager.py

@@ -95,9 +95,13 @@ def run(self):
                 multiChunks = len(node.chunks) > 1
             except TypeError:
                 continue
+            
 
+            print("(tm) A")
             node.preprocess()
+            print("(tm) B")
             for cId, chunk in enumerate(node.chunks):
+                print("(tm) C", cId, chunk)
                 if chunk.isFinishedOrRunning() or not self.isRunning():
                     continue
 
@@ -107,13 +111,17 @@ def run(self):
                 _nodeName, _node, _nbNodes = node.nodeType, nId+1, len(self._manager._nodesToProcess)
 
                 if multiChunks:
+                    print("(tm) D0")
                     _chunk, _nbChunks = cId+1, len(node.chunks)
                     logging.info(f"[{_node}/{_nbNodes}]({_chunk}/{_nbChunks}) {_nodeName}")
                 else:
+                    print("(tm) D1")
                     logging.info(f"[{_node}/{_nbNodes}] {_nodeName}")
                 try:
+                    print("(tm) D")
                     chunk.process(self.forceCompute)
                 except Exception as exc:
+                    print("(tm) D2")
                     if chunk.isStopped():
                         stopAndRestart = True
                         break
@@ -128,6 +136,7 @@ def run(self):
                                 # Node already removed (for instance a global clear of _nodesToProcess)
                                 pass
                             n.clearSubmittedChunks()
+            print("(tm) E")
             node.postprocess()
 
             if stopAndRestart: