move wrapper to csst_common

csst_common/wrapper.py

+import functools
+import time
+import traceback
+from collections import namedtuple
+
+from csst_common.data_manager import CsstMsDataManager
+from csst_common.file_recorder import FileRecorder
+from csst_common.status import CsstStatus
+
+__all__ = ["ModuleResult", "l1ppl_module"]
+
+# module should return ModuleResult as result
+ModuleResult = namedtuple("ModuleResult", ["module", "timecost", "status", "fr", "output"])
+
+
+def l1ppl_module(func):
+    @functools.wraps(func)
+    def call_l1ppl_module(dm: CsstMsDataManager, *args, **kwargs):
+        dm.logger_ppl.info(f"=====================================================")
+        t_start = time.time()
+        dm.logger_ppl.info(f"Starting Module: **{func.__name__}**")
+        # dm.logger_ppl.info(f"Additional arguments: {args} {kwargs}")
+        try:
+            # if the module finishes
+            status, fr, *output = func(dm, *args, **kwargs)
+        except Exception as e:
+            # if the module raises error
+            exc_info = traceback.format_exc()  # traceback info
+            dm.logger_ppl.error(f"Error occurs! \n{exc_info}")
+            status = CsstStatus.ERROR  # default status if exceptions occur
+            fr = FileRecorder()  # default FileRecorder if exceptions occur
+            output = [exc_info, ]  # default output if exceptions occur
+        finally:
+            t_stop = time.time()
+            t_cost = t_stop - t_start
+            if status in [CsstStatus.PERFECT, CsstStatus.WARNING, CsstStatus.ERROR]:
+                # perfect / warning / error
+                dm.logger_ppl.info(f"Module finished with status: {status} - time cost: {t_cost:.1f} sec")
+            else:
+                # invalid status
+                dm.logger_ppl.error(f"Invalid status: {status}")
+            # record exception traceback info
+            dm.logger_ppl.info(
+                f"ModuleResult: \n"
+                f"   - name: {func.__name__}\n"
+                f"   - status: {status}\n"
+                f"   - additional output: {output}\n"
+                f"   - fr: [{fr.summary}]\n{fr}\n"
+            )
+            # write time stamp
+            dm.write_stamp()
+            return ModuleResult(func.__name__, t_cost, status, fr, output)
+
+    return call_l1ppl_module
+
+
+if __name__ == "__main__":
+    @l1ppl_module
+    def call_add(dm, a, b):
+        if isinstance(a, float) and isinstance(b, float):
+            return ModuleResult(CsstStatus.PERFECT, None, a + b)
+        else:
+            return ModuleResult(CsstStatus.ERROR, None, (a, b))
+
+    dm = CsstMsDataManager()
+    print(call_add(dm, 1., 2.))
+    print(call_add(dm, 1., None))