83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
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")
|