refactor: cm_api -> cm_global.APIs

glob/cm_global.py

@@ -0,0 +1,112 @@
+import traceback
+
+#
+# Global Var
+#
+# Usage:
+#   import cm_global
+#   cm_global.variables['comfyui.revision'] = 1832
+#   print(f"log mode: {cm_global.variables['logger.enabled']}")
+#
+variables = {}
+
+
+#
+# Global API
+#
+# Usage:
+# [register API]
+#   import cm_global
+#
+#   def api_hello(msg):
+#       print(f"hello: {msg}")
+#       return msg
+#
+#   cm_global.register_api('hello', api_hello)
+#
+# [use API]
+#   import cm_global
+#
+#   test = cm_global.try_call(api='hello', msg='an example')
+#   print(f"'{test}' is returned")
+#
+
+APIs = {}
+
+
+def register_api(k, f):
+    global APIs
+    APIs[k] = f
+
+
+def try_call(**kwargs):
+    if 'api' in kwargs:
+        api_name = kwargs['api']
+        try:
+            api = APIs.get(api_name)
+            if api is not None:
+                del kwargs['api']
+                return api(**kwargs)
+            else:
+                print(f"WARN: The '{kwargs['api']}' API has not been registered.")
+        except Exception as e:
+            print(f"ERROR: An exception occurred while calling the '{api_name}' API.")
+            raise e
+    else:
+        return None
+
+
+#
+# Extension Info
+#
+# Usage:
+#   import cm_global
+#
+#   cm_global.extension_infos['my_extension'] = {'version': [0, 1], 'name': 'me', 'description': 'example extension', }
+#
+extension_infos = {}
+
+on_extension_registered_handlers = {}
+
+
+def register_extension(extension_name, v):
+    global extension_infos
+    global on_extension_registered_handlers
+    extension_infos[extension_name] = v
+
+    if extension_name in on_extension_registered_handlers:
+        for k, f in on_extension_registered_handlers[extension_name]:
+            try:
+                f(extension_name, v)
+            except Exception:
+                print(f"[ERROR] '{k}' on_extension_registered_handlers")
+                traceback.print_exc()
+
+        del on_extension_registered_handlers[extension_name]
+
+
+def add_on_extension_registered(k, extension_name, f):
+    global on_extension_registered_handlers
+    if extension_name in extension_infos:
+        try:
+            v = extension_infos[extension_name]
+            f(extension_name, v)
+        except Exception:
+            print(f"[ERROR] '{k}' on_extension_registered_handler")
+            traceback.print_exc()
+    else:
+        if extension_name not in on_extension_registered_handlers:
+            on_extension_registered_handlers[extension_name] = []
+
+        on_extension_registered_handlers[extension_name].append((k, f))
+
+
+def add_on_revision_detected(k, f):
+    if 'comfyui.revision' in variables:
+        try:
+            f(variables['comfyui.revision'])
+        except Exception:
+            print(f"[ERROR] '{k}' on_revision_detected_handler")
+            traceback.print_exc()
+    else:
+        variables['cm.on_revision_detected_handler'].append((k, f))