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

from __future__ import annotations

from typing import Any, Iterable

from wandb.proto import wandb_settings_pb2
from wandb.sdk.lib import RunMoment
from wandb.sdk.wandb_settings import Settings


class SettingsStatic(Settings):
    """A readonly object that wraps a protobuf Settings message.

    Implements the mapping protocol, so you can access settings as
    attributes or items.
    """

    def __init__(self, proto: wandb_settings_pb2.Settings) -> None:
        data = self._proto_to_dict(proto)
        super().__init__(**data)

    def _proto_to_dict(self, proto: wandb_settings_pb2.Settings) -> dict:
        data = {}

        exclude_fields = {
            "model_config",
            "model_fields",
            "model_fields_set",
            "__fields__",
            "__model_fields_set",
            "__pydantic_self__",
            "__pydantic_initialised__",
        }

        fields = (
            Settings.model_fields
            if hasattr(Settings, "model_fields")
            else Settings.__fields__
        )  # type: ignore [attr-defined]

        fields = {k: v for k, v in fields.items() if k not in exclude_fields}  # type: ignore [union-attr]

        forks_specified: list[str] = []
        for key in fields:
            # Skip Python-only keys that do not exist on the proto.
            if key in ("reinit",):
                continue

            value: Any = None

            field_info = fields[key]
            annotation = str(field_info.annotation)

            if key == "_stats_open_metrics_filters":
                # todo: it's an underscored field, refactor into
                #  something more elegant?
                # I'm really about this. It's ugly, but it works.
                # Do not try to repeat this at home.
                value_type = getattr(proto, key).WhichOneof("value")
                if value_type == "sequence":
                    value = list(getattr(proto, key).sequence.value)
                elif value_type == "mapping":
                    unpacked_mapping = {}
                    for outer_key, outer_value in getattr(
                        proto, key
                    ).mapping.value.items():
                        unpacked_inner = {}
                        for inner_key, inner_value in outer_value.value.items():
                            unpacked_inner[inner_key] = inner_value
                        unpacked_mapping[outer_key] = unpacked_inner
                    value = unpacked_mapping
            elif key == "fork_from" or key == "resume_from":
                value = getattr(proto, key)
                if value.run:
                    value = RunMoment(
                        run=value.run, value=value.value, metric=value.metric
                    )
                    forks_specified.append(key)
                else:
                    value = None
            else:
                if proto.HasField(key):  # type: ignore [arg-type]
                    value = getattr(proto, key).value
                    # Convert to list if the field is a sequence
                    if any(t in annotation for t in ("tuple", "Sequence", "list")):
                        value = list(value)
                else:
                    value = None

            if value is not None:
                data[key] = value

        if len(forks_specified) > 1:
            raise ValueError(
                "Only one of fork_from or resume_from can be specified, not both"
            )

        return data

    def __setattr__(self, name: str, value: object) -> None:
        raise AttributeError("Error: SettingsStatic is a readonly object")

    def __setitem__(self, key: str, val: object) -> None:
        raise AttributeError("Error: SettingsStatic is a readonly object")

    def keys(self) -> Iterable[str]:
        return self.__dict__.keys()

    def __getitem__(self, key: str) -> Any:
        return self.__dict__[key]

    def __getattr__(self, name: str) -> Any:
        try:
            return self.__dict__[name]
        except KeyError:
            raise AttributeError(f"SettingsStatic has no attribute {name}")

    def __str__(self) -> str:
        return str(self.__dict__)

    def __contains__(self, key: str) -> bool:
        return key in self.__dict__