venv/lib/python3.11/site-packages/wandb/sdk/wandb_summary.py

import abc
import typing as t

from .interface.summary_record import SummaryItem, SummaryRecord


def _get_dict(d):
    if isinstance(d, dict):
        return d
    # assume argparse Namespace
    return vars(d)


class SummaryDict(metaclass=abc.ABCMeta):
    """dict-like wrapper for the nested dictionaries in a SummarySubDict.

    Triggers self._root._callback on property changes.
    """

    @abc.abstractmethod
    def _as_dict(self):
        raise NotImplementedError

    @abc.abstractmethod
    def _update(self, record: SummaryRecord):
        raise NotImplementedError

    def keys(self):
        return [k for k in self._as_dict().keys() if k != "_wandb"]

    def get(self, key, default=None):
        return self._as_dict().get(key, default)

    def __getitem__(self, key):
        item = self._as_dict()[key]

        if isinstance(item, dict):
            # this nested dict needs to be wrapped:
            wrapped_item = SummarySubDict()
            object.__setattr__(wrapped_item, "_items", item)
            object.__setattr__(wrapped_item, "_parent", self)
            object.__setattr__(wrapped_item, "_parent_key", key)

            return wrapped_item

        # this item isn't a nested dict
        return item

    __getattr__ = __getitem__

    def __setitem__(self, key, val):
        self.update({key: val})

    __setattr__ = __setitem__

    def __delattr__(self, key):
        record = SummaryRecord()
        item = SummaryItem()
        item.key = (key,)
        record.remove = (item,)
        self._update(record)

    __delitem__ = __delattr__

    def update(self, d: t.Dict):
        # import ipdb; ipdb.set_trace()
        record = SummaryRecord()
        for key, value in d.items():
            item = SummaryItem()
            item.key = (key,)
            item.value = value
            record.update.append(item)

        self._update(record)


class Summary(SummaryDict):
    """Track single values for each metric for each run.

    By default, a metric's summary is the last value of its History.

    For example, `wandb.log({'accuracy': 0.9})` will add a new step to History and
    update Summary to the latest value. In some cases, it's more useful to have
    the maximum or minimum of a metric instead of the final value. You can set
    history manually `(wandb.summary['accuracy'] = best_acc)`.

    In the UI, summary metrics appear in the table to compare across runs.
    Summary metrics are also used in visualizations like the scatter plot and
    parallel coordinates chart.

    After training has completed, you may want to save evaluation metrics to a
    run. Summary can handle numpy arrays and PyTorch/TensorFlow tensors. When
    you save one of these types to Summary, we persist the entire tensor in a
    binary file and store high level metrics in the summary object, such as min,
    mean, variance, and 95th percentile.

    Examples:
        ```python
        wandb.init(config=args)

        best_accuracy = 0
        for epoch in range(1, args.epochs + 1):
            test_loss, test_accuracy = test()
            if test_accuracy > best_accuracy:
                wandb.run.summary["best_accuracy"] = test_accuracy
                best_accuracy = test_accuracy
        ```
    """

    _update_callback: t.Callable
    _get_current_summary_callback: t.Callable

    def __init__(self, get_current_summary_callback: t.Callable):
        super().__init__()
        object.__setattr__(self, "_update_callback", None)
        object.__setattr__(
            self, "_get_current_summary_callback", get_current_summary_callback
        )

    def _set_update_callback(self, update_callback: t.Callable):
        object.__setattr__(self, "_update_callback", update_callback)

    def _as_dict(self):
        return self._get_current_summary_callback()

    def _update(self, record: SummaryRecord):
        if self._update_callback:  # type: ignore
            self._update_callback(record)


class SummarySubDict(SummaryDict):
    """Non-root node of the summary data structure.

    Contains a path to itself from the root.
    """

    _items: t.Dict
    _parent: SummaryDict
    _parent_key: str

    def __init__(self):
        object.__setattr__(self, "_items", dict())
        object.__setattr__(self, "_parent", None)
        object.__setattr__(self, "_parent_key", None)

    def _as_dict(self):
        return self._items

    def _update(self, record: SummaryRecord):
        return self._parent._update(record._add_next_parent(self._parent_key))