1  import json
 2  from typing import Any
 3  from typing import Callable
 4  from typing import Iterator
 5  from typing import List
 6  from typing import Optional
 7  from typing import Tuple
 8  
 9  from evidently._pydantic_compat import BaseModel
10  from evidently.legacy.base_metric import MetricResult
11  from evidently.legacy.core import BaseResult
12  from evidently.legacy.utils import NumpyEncoder
13  
14  
15  def iterate_obj_fields(
16      obj: Any, paths: List[str], early_stop: Optional[Callable[[Any, List[str]], Optional[List[Tuple[str, Any]]]]] = None
17  ) -> Iterator[Tuple[str, Any]]:
18      if early_stop is not None:
19          es = early_stop(obj, paths)
20          if es is not None:
21              yield from es
22              return
23  
24      from evidently.ui.backport import ByLabelCountValueV1
25      from evidently.ui.backport import ByLabelValueV1
26  
27      if isinstance(obj, ByLabelValueV1):
28          yield from (
29              [(".".join(paths + ["values"]), obj.values)]
30              + [(".".join(paths + ["values", str(key)]), str(val)) for key, val in obj.values.items()]
31          )
32          return
33      if isinstance(obj, ByLabelCountValueV1):
34          yield from (
35              [(".".join(paths + ["counts"]), obj.counts)]
36              + [(".".join(paths + ["counts", str(key)]), str(val)) for key, val in obj.counts.items()]
37              + [(".".join(paths + ["shares"]), obj.shares)]
38              + [(".".join(paths + ["shares", str(key)]), str(val)) for key, val in obj.shares.items()]
39          )
40          return
41      if isinstance(obj, list):
42          return
43      if isinstance(obj, dict):
44          yield from (r for key, value in obj.items() for r in iterate_obj_fields(value, paths + [str(key)], early_stop))
45          return
46      if isinstance(obj, BaseResult) and obj.__config__.extract_as_obj:
47          yield ".".join(paths), obj
48          return
49      if isinstance(obj, BaseModel):
50          yield from (
51              r
52              for name, field in obj.__fields__.items()
53              for r in iterate_obj_fields(getattr(obj, name), paths + [name], early_stop)
54          )
55          return
56      yield ".".join(paths), obj
57  
58  
59  def iterate_obj_float_fields(obj: Any, paths: List[str]) -> Iterator[Tuple[str, str]]:
60      for path, value in iterate_obj_fields(obj, paths):
61          if isinstance(value, dict):
62              yield path, json.dumps(value, cls=NumpyEncoder)
63              continue
64          if isinstance(value, BaseResult) and value.__config__.extract_as_obj:
65              yield path, json.dumps(value.dict(), cls=NumpyEncoder)
66              continue
67          try:
68              value = str(float(value))
69              yield path, value
70          except (TypeError, ValueError):
71              continue
72  
73  
74  def iterate_metric_results_fields(metric_result: MetricResult) -> Iterator[Tuple[str, str]]:
75      yield from iterate_obj_float_fields(metric_result, [])