dify init

core/metric_manager.py

@@ -0,0 +1,82 @@
+from pathlib import Path
+
+
+class MetricManager:
+    def __init__(self, metric):
+        self.metric = metric
+
+    def new_span(self, m_type: str = "default", m_id: str = "default", m_iter: int = 1):
+        self.metric.span(m_type, m_id, m_iter)
+
+    def _smart_name_label(self, name: str, label: str | None) -> tuple[str, str]:
+        """retruns: (name,label)"""
+        if not name:
+            raise ValueError()
+        if not name.strip():
+            raise ValueError("Empty name")
+
+        if label:
+            return name, label
+        for sep in ("|", ":"):
+            if sep in name:
+                _n, _l = name.split(sep, maxsplit=1)
+                return (_n, _l)
+        return name, name
+
+    def add_text_metric(self, name: str, value: str, label: str | None = None):
+        _n, _l = self._smart_name_label(name, label)
+        self.metric.add(
+            name=_n,
+            label=_l,
+            value=value,
+            type="text",
+        )
+
+    def add_number_metric(self, name: str, value: float, label: str | None = None):
+        _n, _l = self._smart_name_label(name, label)
+        self.metric.add(
+            name=_n,
+            label=_l,
+            value=value,
+            type="number",
+        )
+
+    def add_image_metric(self, name: str, value: Path | str, label: str | None = None):
+        _n, _l = self._smart_name_label(name, label)
+        self.metric.add(
+            name=_n,
+            label=_l,
+            value=str(value),
+            type="image",
+        )
+
+    def add_video_metric(self, name: str, value: Path | str, label: str | None = None):
+        _n, _l = self._smart_name_label(name, label)
+        self.metric.add(
+            name=_n,
+            label=_l,
+            value=value,
+            type="video",
+        )
+
+    def send(self):
+        self.metric.send_all()
+
+    def from_dict(self, the_dict: dict):
+        if not the_dict:
+            raise ValueError("dict is null")
+        for k, v in the_dict.items():
+            if isinstance(v, (int, float)):
+                self.add_number_metric(name=k, value=v)
+            elif isinstance(v, str):
+                self.add_text_metric(name=k, value=v)
+            elif isinstance(v, Path):
+                if v.suffix.lower() in (".jpg", ".png", ".gif", ".bmp"):
+                    self.add_image_metric(name=k, value=str(v.resolve()))
+                elif v.suffix.lower() in (".mp4", ".mkv"):
+                    self.add_video_metric(name=k, value=str(v.resolve()))
+                else:
+                    self.add_text_metric(name=k, value=str(v.resolve()))
+
+            else:
+                raise NotImplementedError(f"{type(v)} not supported")