Merge pull request #384 from arayabrain/feature/cloud/improve-post-pr…

…ocess Feature/cloud/improve post process
arayabrain · Jul 16, 2024 · 7c20b5d · 7c20b5d
2 parents 6dae402 + c01f50d
commit 7c20b5d
Show file tree

Hide file tree

Showing 43 changed files with 356 additions and 79 deletions.
diff --git a/frontend/build/asset-manifest.json b/frontend/build/asset-manifest.json
@@ -1,16 +1,16 @@
 {
   "files": {
     "main.css": "/static/css/main.341bf264.css",
-    "main.js": "/static/js/main.7a24960b.js",
+    "main.js": "/static/js/main.9adb5116.js",
     "static/js/47.145ba9bc.chunk.js": "/static/js/47.145ba9bc.chunk.js",
     "static/media/logo.png": "/static/media/logo.e44efe3246aecc6d6219.png",
     "index.html": "/index.html",
     "main.341bf264.css.map": "/static/css/main.341bf264.css.map",
-    "main.7a24960b.js.map": "/static/js/main.7a24960b.js.map",
+    "main.9adb5116.js.map": "/static/js/main.9adb5116.js.map",
     "47.145ba9bc.chunk.js.map": "/static/js/47.145ba9bc.chunk.js.map"
   },
   "entrypoints": [
     "static/css/main.341bf264.css",
-    "static/js/main.7a24960b.js"
+    "static/js/main.9adb5116.js"
   ]
 }
diff --git a/frontend/build/index.html b/frontend/build/index.html
@@ -1 +1 @@
-<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/static/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Web site created using create-react-app"/><link rel="manifest" href="/static/manifest.json"/><title>OptiNiSt</title><script defer="defer" src="/static/js/main.7a24960b.js"></script><link href="/static/css/main.341bf264.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>
+<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/static/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Web site created using create-react-app"/><link rel="manifest" href="/static/manifest.json"/><title>OptiNiSt</title><script defer="defer" src="/static/js/main.9adb5116.js"></script><link href="/static/css/main.341bf264.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>
diff --git a/frontend/build/static/js/main.7a24960b.js → frontend/build/static/js/main.9adb5116.js b/frontend/build/static/js/main.7a24960b.js → frontend/build/static/js/main.9adb5116.js
diff --git a/...ld/static/js/main.7a24960b.js.LICENSE.txt → ...ld/static/js/main.9adb5116.js.LICENSE.txt b/...ld/static/js/main.7a24960b.js.LICENSE.txt → ...ld/static/js/main.9adb5116.js.LICENSE.txt
diff --git a/...tend/build/static/js/main.7a24960b.js.map → ...tend/build/static/js/main.9adb5116.js.map b/...tend/build/static/js/main.7a24960b.js.map → ...tend/build/static/js/main.9adb5116.js.map
diff --git a/frontend/src/components/Workspace/Experiment/Experiment.tsx b/frontend/src/components/Workspace/Experiment/Experiment.tsx
@@ -5,9 +5,11 @@ import Loading from "components/common/Loading"
 import { ExperimentTable } from "components/Workspace/Experiment/ExperimentTable"
 import { CONTENT_HEIGHT } from "const/Layout"
 import { selectLoading } from "store/slice/Experiments/ExperimentsSelectors"
+import { selectLoading as selectFlowElementLoading } from "store/slice/FlowElement/FlowElementSelectors"
 
 const Experiment = memo(function Experiment() {
   const loading = useSelector(selectLoading)
+  const flowElementLoading = useSelector(selectFlowElementLoading)
 
   return (
     <div style={{ display: "flex" }}>
@@ -20,7 +22,7 @@ const Experiment = memo(function Experiment() {
         }}
       >
         <ExperimentTable />
-        <Loading loading={loading} />
+        <Loading loading={loading || flowElementLoading} />
       </main>
     </div>
   )

diff --git a/frontend/src/store/slice/FlowElement/FlowElementSlice.ts b/frontend/src/store/slice/FlowElement/FlowElementSlice.ts
@@ -223,7 +223,13 @@ export const flowElementSlice = createSlice({
       .addCase(fetchWorkflow.pending, (state) => {
         state.loading = true
       })
+      .addCase(reproduceWorkflow.pending, (state) => {
+        state.loading = true
+      })
       .addCase(fetchWorkflow.rejected, () => initialState)
+      .addCase(reproduceWorkflow.rejected, (state) => {
+        state.loading = false
+      })
       .addMatcher(
         isAnyOf(
           reproduceWorkflow.fulfilled,

diff --git a/frontend/src/store/slice/Pipeline/PipelineType.ts b/frontend/src/store/slice/Pipeline/PipelineType.ts
@@ -117,3 +117,10 @@ export const RUN_BTN_LABELS = {
   [RUN_BTN_OPTIONS.RUN_NEW]: "RUN ALL",
   [RUN_BTN_OPTIONS.RUN_ALREADY]: "RUN",
 } as const
+
+export const PROCESS_TYPE = {
+  POST_PROCESS: {
+    id: "post_process_0",
+    name: "post_process",
+  },
+} as const
diff --git a/frontend/src/store/slice/Pipeline/PipelineUtils.ts b/frontend/src/store/slice/Pipeline/PipelineUtils.ts
@@ -12,6 +12,7 @@ import {
   NodeResultError,
   NodeResultSuccess,
   NodeResult,
+  PROCESS_TYPE,
 } from "store/slice/Pipeline/PipelineType"
 
 export function isNodeResultPending(
@@ -52,6 +53,14 @@ export function getInitialRunResult(runPostData: RunPostData) {
         name: data?.label ?? "",
       }
     })
+
+  // NOTE: Force the addition of a parameter here
+  //     to confirm post_process processing.
+  initialResult[PROCESS_TYPE.POST_PROCESS.id] = {
+    status: NODE_RESULT_STATUS.PENDING,
+    name: PROCESS_TYPE.POST_PROCESS.name,
+  }
+
   return initialResult
 }
 

diff --git a/studio/app/Snakefile b/studio/app/Snakefile
@@ -33,7 +33,7 @@ else:
 
         # make rule of post_process
         elif details["type"] in [
-            ProcessType.POST_PROCESS,
+            ProcessType.POST_PROCESS.type,
         ]:
             rule:
                 name:

diff --git a/studio/app/common/core/experiment/experiment.py b/studio/app/common/core/experiment/experiment.py
@@ -28,5 +28,28 @@ class ExptConfig:
     success: Optional[str]
     hasNWB: bool
     function: Dict[str, ExptFunction]
+    procs: Optional[Dict[str, ExptFunction]]
     nwb: NWBParams
     snakemake: SmkParam
+
+
+@dataclass
+class ExptOutputPathIds:
+    output_dir: Optional[str] = None
+    workspace_id: Optional[str] = None
+    unique_id: Optional[str] = None
+    function_id: Optional[str] = None
+
+    def __post_init__(self):
+        """
+        Extract each ID from output_path
+        - output_dir format
+          - {DIRPATH.OUTPUT_DIR}/{workspace_id}/{unique_id}/{function_id}
+        """
+        if self.output_dir:
+            splitted_path = self.output_dir.split("/")
+        else:
+            splitted_path = []
+
+        if len(splitted_path) >= 3:
+            self.workspace_id, self.unique_id, self.function_id = splitted_path[-3:]
diff --git a/studio/app/common/core/experiment/experiment_builder.py b/studio/app/common/core/experiment/experiment_builder.py
@@ -11,6 +11,7 @@ def __init__(self):
         self._success = None
         self._hasNWB = False
         self._function = {}
+        self._procs = {}
         self._nwbfile = None
         self._snakemake = None
 
@@ -23,6 +24,7 @@ def set_config(self, config: ExptConfig) -> "ExptConfigBuilder":
         self._success = config.success
         self._hasNWB = config.hasNWB
         self._function = config.function
+        self._procs = config.procs
         self._nwbfile = config.nwb
         self._snakemake = config.snakemake
         return self
@@ -55,6 +57,10 @@ def set_function(self, function) -> "ExptConfigBuilder":
         self._function = function
         return self
 
+    def set_procs(self, function) -> "ExptConfigBuilder":
+        self._procs = function
+        return self
+
     def set_nwbfile(self, nwbfile) -> "ExptConfigBuilder":
         self._nwbfile = nwbfile
         return self
@@ -73,6 +79,7 @@ def build(self) -> ExptConfig:
             success=self._success,
             hasNWB=self._hasNWB,
             function=self._function,
+            procs=self._procs,
             nwb=self._nwbfile,
             snakemake=self._snakemake,
         )
diff --git a/studio/app/common/core/experiment/experiment_reader.py b/studio/app/common/core/experiment/experiment_reader.py
@@ -21,12 +21,18 @@ def read(cls, filepath) -> ExptConfig:
             success=config.get("success", "running"),
             hasNWB=config["hasNWB"],
             function=cls.read_function(config["function"]),
+            procs=cls.read_function(config.get("procs")),
             nwb=config.get("nwb"),
             snakemake=config.get("snakemake"),
         )
 
     @classmethod
     def read_function(cls, config) -> Dict[str, ExptFunction]:
+        # Assuming the case where an empty value is specified, check here.
+        # (For backward compatibility with config yaml)
+        if not config:
+            return {}
+
         return {
             key: ExptFunction(
                 unique_id=value["unique_id"],

diff --git a/studio/app/common/core/experiment/experiment_writer.py b/studio/app/common/core/experiment/experiment_writer.py
@@ -11,6 +11,7 @@
 from studio.app.common.core.experiment.experiment_reader import ExptConfigReader
 from studio.app.common.core.utils.config_handler import ConfigWriter
 from studio.app.common.core.utils.filepath_creater import join_filepath
+from studio.app.common.core.workflow.workflow import ProcessType
 from studio.app.common.core.workflow.workflow_reader import WorkflowConfigReader
 from studio.app.const import DATE_FORMAT
 from studio.app.dir_path import DIRPATH
@@ -48,7 +49,8 @@ def write(self) -> None:
         else:
             self.create_config()
 
-        self.function_from_nodeDict()
+        self.build_function_from_nodeDict()
+        self.build_procs()
 
         ConfigWriter.write(
             dirname=join_filepath(
@@ -77,7 +79,7 @@ def add_run_info(self) -> ExptConfig:
             .build()
         )
 
-    def function_from_nodeDict(self) -> ExptConfig:
+    def build_function_from_nodeDict(self) -> ExptConfig:
         func_dict: Dict[str, ExptFunction] = {}
         node_dict = WorkflowConfigReader.read(
             join_filepath(
@@ -102,6 +104,17 @@ def function_from_nodeDict(self) -> ExptConfig:
 
         return self.builder.set_function(func_dict).build()
 
+    def build_procs(self) -> ExptConfig:
+        target_procs = [ProcessType.POST_PROCESS]
+        func_dict: Dict[str, ExptFunction] = {}
+
+        for proc in target_procs:
+            func_dict[proc.id] = ExptFunction(
+                unique_id=proc.id, name=proc.label, hasNWB=False, success="running"
+            )
+
+        return self.builder.set_procs(func_dict).build()
+
 
 class ExptDataWriter:
     def __init__(
@@ -143,6 +156,7 @@ def rename(self, new_name: str) -> ExptConfig:
             success=config.get("success", "running"),
             hasNWB=config["hasNWB"],
             function=ExptConfigReader.read_function(config["function"]),
+            procs=ExptConfigReader.read_function(config.get("procs")),
             nwb=config.get("nwb"),
             snakemake=config.get("snakemake"),
         )
diff --git a/studio/app/common/core/rules/post_process.py b/studio/app/common/core/rules/post_process.py
@@ -6,15 +6,18 @@
 import json
 import os
 import sys
+import traceback
 from os.path import abspath, dirname
 from pathlib import Path
 
 ROOT_DIRPATH = dirname(dirname(dirname(dirname(dirname(dirname(abspath(__file__)))))))
 
 sys.path.append(ROOT_DIRPATH)
 
+from studio.app.common.core.experiment.experiment import ExptOutputPathIds
 from studio.app.common.core.logger import AppLogger
 from studio.app.common.core.rules.runner import Runner
+from studio.app.common.core.snakemake.smk import Rule
 from studio.app.common.core.snakemake.snakemake_reader import RuleConfigReader
 from studio.app.common.core.storage.remote_storage_controller import (
     RemoteStorageController,
@@ -26,35 +29,64 @@
 
 logger = AppLogger.get_logger()
 
+
+class PostProcessRunner:
+    @classmethod
+    def run(cls, __rule: Rule):
+        try:
+            logger.info("start post_process runner")
+
+            # Get input data for a rule.
+            # Note:
+            #   - Check if all node data can be successfully retrieved.
+            #     If there is an error in any node, an AssertionError is generated here.
+            #   - read_input_info() is used to determine if there is an error,
+            #     and the return value is not used here.
+            Runner.read_input_info(__rule.input)
+
+            # Operate remote storage.
+            if RemoteStorageController.use_remote_storage():
+                # Get workspace_id, unique_id from output file path
+                ids = ExptOutputPathIds(dirname(__rule.output))
+                workspace_id = ids.workspace_id
+                unique_id = ids.unique_id
+
+                # 処理結果データの外部ストレージへのデータ転送
+                # TODO: データ転送の実施要否の判定追加（環境変数想定）
+                remote_storage_controller = RemoteStorageController()
+                remote_storage_controller.upload_experiment(workspace_id, unique_id)
+            else:
+                logger.debug("remote storage is unused in post_process.")
+
+            # 処理結果ファイルを保存
+            output_path = __rule.output
+            output_info = {"success": True}  # TODO: より適切な情報の記録に修正想定
+            PickleWriter.write(output_path, output_info)
+
+        except Exception as e:
+            """
+            Note: The code here is the same as in the except section of Runner.run
+            """
+            err_msg = list(traceback.TracebackException.from_exception(e).format())
+
+            # show full trace to console
+            logger.error("\n".join(err_msg))
+
+            # save msg for GUI
+            PickleWriter.write(__rule.output, err_msg)
+
+
 if __name__ == "__main__":
     # TODO: debug log.
     logger.debug(
-        ">>>>>>>>> Post process logging\n"
+        ">>>>>>>>>>>>>>>>>>>> Post process logging\n"
         f"[snakemake.input: {snakemake.input}]\n"
         f"[snakemake.output: {snakemake.output}]\n"
     )
 
-    # Get input data for a rule.
-    # Note: Check if all node data can be successfully retrieved.
-    #       If there is an error in any node, an AssertionError is generated here.
-    input_info = Runner.read_input_info(snakemake.input)
-    del input_info
-
-    # Operate remote storage.
-    if RemoteStorageController.use_remote_storage():
-        # Get workspace_id, unique_id
-        # TODO: 関数（Utility）化
-        rule_config = RuleConfigReader.read(snakemake.params.name)
-        workspace_id, unique_id = rule_config.output.split("/")[0:2]
-
-        # 処理結果データの外部ストレージへのデータ転送
-        # TODO: データ転送の実施要否の判定追加（環境変数想定）
-        remote_storage_controller = RemoteStorageController()
-        remote_storage_controller.upload_experiment(workspace_id, unique_id)
-    else:
-        logger.debug("remote storage is unused in post-process.")
-
-    # 処理結果ファイルを保存
-    output_path = str(snakemake.output)
-    output_info = {"success": True}  # TODO: より適切な情報の記録に要修正
-    PickleWriter.write(output_path, output_info)
+    rule_config = RuleConfigReader.read(snakemake.params.name)
+
+    rule_config.input = snakemake.input
+    rule_config.output = snakemake.output[0]
+
+    PostProcessRunner.run(rule_config)
diff --git a/studio/app/common/core/snakemake/smk.py b/studio/app/common/core/snakemake/smk.py
@@ -23,6 +23,12 @@ class FlowConfig:
     last_output: list
 
 
+class NormalRun(BaseModel):
+    id: str
+    type: str
+    label: str
+
+
 class ForceRun(BaseModel):
     nodeId: str
     name: str

diff --git a/studio/app/common/core/snakemake/smk_utils.py b/studio/app/common/core/snakemake/smk_utils.py
@@ -30,7 +30,7 @@ def conda(cls, details):
             return None
         # skip conda for process-event
         elif details["type"] in [
-            ProcessType.POST_PROCESS,
+            ProcessType.POST_PROCESS.type,
         ]:
             return None