Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
193 changes: 108 additions & 85 deletions cyclops/report/model_card/base.py
Original file line number Diff line number Diff line change
@@ -1,100 +1,130 @@
"""Base classes for model card fields and sections."""

import keyword
from typing import Any, Callable, Dict, List, Literal, Optional, Union
import typing
from typing import Any, ClassVar, Dict, List, Literal, Optional, Union

import numpy as np
from pydantic import BaseConfig, BaseModel, Extra, root_validator
from pydantic.fields import FieldInfo, ModelField
from pydantic import BaseModel, ConfigDict, model_validator
from pydantic.fields import FieldInfo


def _unwrap_optional(annotation: Any) -> Any:
"""Unwrap `Optional[SomeType]`/`Union[SomeType, None]` to `SomeType`."""
if typing.get_origin(annotation) is Union:
args = [arg for arg in typing.get_args(annotation) if arg is not type(None)]
if len(args) == 1:
return args[0]
return annotation


def unwrap_optional_type(field_info: FieldInfo) -> Any:
"""Get the annotated type of a field, unwrapping `Optional`/`Union[..., None]`.

All model card fields/sections are declared as `Optional[SomeType]`, so
`field_info.annotation` is `Union[SomeType, None]`. This returns `SomeType`.
"""
return _unwrap_optional(field_info.annotation)


def _get_list_item_type(field_info: FieldInfo) -> Any:
"""Get the item type of a `List[X]`/`Optional[List[X]]` field annotation."""
annotation = _unwrap_optional(field_info.annotation)
if typing.get_origin(annotation) is list:
args = typing.get_args(annotation)
if args:
return args[0]
return annotation


def _check_composable_fields(
cls: BaseModel,
cls: type,
values: Dict[str, Any],
) -> Dict[str, Any]:
"""Check that the type of the field is allowed in the section."""
for attr, value in values.items():
if issubclass(type(value), BaseModelCardField) and (
value.__config__.composable_with is None
value.composable_with is None
or (
value.__config__.composable_with != "Any"
and cls.__name__ not in value.__config__.composable_with # type: ignore [attr-defined] # noqa: E501
value.composable_with != "Any"
and cls.__name__ not in value.composable_with
)
):
print(issubclass(type(value), BaseModelCardField))
print(value.__config__.composable_with)
raise ValueError(
f"Field `{attr}`(type={type(value)}) is not allowed in "
f"`{cls.__name__}` section.", # type: ignore[attr-defined]
f"`{cls.__name__}` section.",
)

return values


class BaseModelCardConfig(BaseConfig):
"""Global config for model card fields.

Attributes
----------
extra : Extra
Whether to allow extra attributes in the field object.
smart_union : bool
Whether `Union` should check all allowed types before even trying to
coerce.
validate_all : bool
Whether to validate all attributes in the field object.
validate_assignment : bool
Whether to validate assignments of attributes in the field object.
json_encoders : Dict[Any, Callable]
Custom JSON encoders.

"""

extra: Extra = Extra.allow
smart_union: bool = True
validate_all: bool = True
validate_assignment: bool = True
json_encoders: Dict[Any, Callable[..., Any]] = {np.ndarray: lambda v: v.tolist()}
# Shared config for model card fields and sections.
# - extra: allow extra attributes on the field/section object.
# - validate_default: validate default values.
# - validate_assignment: validate attribute assignments.
# - json_encoders: custom JSON encoders.
_MODEL_CARD_CONFIG = ConfigDict(
extra="allow",
validate_default=True,
validate_assignment=True,
json_encoders={np.ndarray: lambda v: v.tolist()},
)


class BaseModelCardField(BaseModel):
"""Base class for model card fields."""
"""Base class for model card fields.

Class Parameters
----------------
composable_with : Literal["Any"], List[str], optional
The sections this field can be dynamically composed with. If "Any",
the field can be composed with any subclass of `BaseModelCardSection`
or `BaseModelCardField`. If None, the field cannot be dynamically
composed with any other fields or sections - it must be explicitly
added to a section. If a list of strings, the strings are the names
treated as class names for subclasses of `BaseModelCardSection` or
`BaseModelCardField`.
list_factory : bool, default=False
Whether multiple instances of this field can be added to a section
in a list.

class Config(BaseModelCardConfig):
"""Global config for model card fields.
"""

Attributes
----------
composable_with : Literal["Any"], List[str], optional
The sections this field can be dynamically composed with. If "Any",
the field can be composed with any subclass of `BaseModelCardSection`
or `BaseModelCardField`. If None, the field cannot be dynamically
composed with any other fields or sections - it must be explicitly
added to a section. If a list of strings, the strings are the names
treated as class names for subclasses of `BaseModelCardSection` or
`BaseModelCardField`.
list_factory : bool, default=False
Whether multiple instances of this field can be added to a section
in a list.
composable_with: ClassVar[Optional[Union[Literal["Any"], List[str]]]] = "Any"
list_factory: ClassVar[bool] = False

"""
model_config = _MODEL_CARD_CONFIG

composable_with: Optional[Union[Literal["Any"], List[str]]] = "Any"
list_factory: bool = False
def __init_subclass__(
cls,
composable_with: Optional[Union[Literal["Any"], List[str]]] = "Any",
list_factory: bool = False,
**kwargs: Any,
) -> None:
"""Capture the `composable_with`/`list_factory` class arguments."""
super().__init_subclass__(**kwargs)
cls.composable_with = composable_with
cls.list_factory = list_factory

_validate_composition = root_validator(pre=True, allow_reuse=True)(
_check_composable_fields,
)
@model_validator(mode="before")
@classmethod
def _validate_composition(cls, values: Any) -> Any:
if isinstance(values, dict):
return _check_composable_fields(cls, values)
return values


class BaseModelCardSection(BaseModel):
"""Base class for model card sections."""

Config = BaseModelCardConfig
model_config = _MODEL_CARD_CONFIG

_validate_composition = root_validator(pre=True, allow_reuse=True)(
_check_composable_fields,
)
@model_validator(mode="before")
@classmethod
def _validate_composition(cls, values: Any) -> Any:
if isinstance(values, dict):
return _check_composable_fields(cls, values)
return values

def update_field(self, name: str, value: Any) -> None:
"""Update the field with the given name to the given value.
Expand All @@ -114,11 +144,23 @@ def update_field(self, name: str, value: Any) -> None:
If the field does not exist.

"""
if name not in self.__fields__:
if name not in type(self).model_fields:
raise ValueError(f"Field {name} does not exist.")

field = self.__fields__[name]
field = type(self).model_fields[name]
if field.default_factory == list or isinstance(getattr(self, name), list): # noqa: E721
item_type = _get_list_item_type(field)
if (
isinstance(value, BaseModel)
and isinstance(item_type, type)
and not isinstance(value, item_type)
):
# pydantic v2 does not automatically coerce an arbitrary
# BaseModel instance into a differently-typed nested model
# (unlike pydantic v1); dump it to a dict first so the list
# validation below can construct the expected item type.
value = value.model_dump()

# NOTE: pydantic does not trigger validation when appending to a list,
# but if `validate_assignment` is set to `True`, then validation will
# be triggered when the list is assigned to the field.
Expand Down Expand Up @@ -150,29 +192,10 @@ def add_field(self, name: str, value: Any) -> None:
f" Got {name} instead.",
)

type_ = type(value)
default_factory = None
if isinstance(value, list):
default_factory = list
if (
isinstance(value, BaseModelCardField)
and value.__config__.list_factory is True # type: ignore[attr-defined]
):
default_factory = list
if isinstance(value, BaseModelCardField) and value.list_factory is True:
value = [value] # add field as a list
type_ = List[type_] # type: ignore[valid-type]

# with `extra="allow"`, assigning a name that is not a declared model
# field stores it as an extra attribute, which is included in
# `model_dump`/`model_dump_json` output like any other field
setattr(self, name, value)

# modify __fields__ to include new field
self.__fields__[name] = ModelField(
name=name,
type_=type_,
required=False,
class_validators=None,
model_config=BaseModelCardField.Config,
default_factory=default_factory,
field_info=FieldInfo(unique_items=True)
if default_factory == list # noqa: E721
else None,
)
53 changes: 29 additions & 24 deletions cyclops/report/model_card/fields.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import inspect
from datetime import date as dt_date
from datetime import datetime as dt_datetime
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
from typing import Any, List, Literal, Optional, Tuple, Union

import numpy as np
import numpy.typing as npt
Expand All @@ -17,8 +17,8 @@
StrictFloat,
StrictInt,
StrictStr,
root_validator,
validator,
field_validator,
model_validator,
)

from cyclops.report.model_card.base import BaseModelCardField
Expand Down Expand Up @@ -90,24 +90,33 @@ class License(
)
text_url: Optional[AnyUrl] = Field(None, description="A URL to the license text.")

@root_validator(skip_on_failure=True)
def validate_spdx_identifier(
cls: "License", # noqa: N805
values: Dict[str, StrictStr],
) -> Dict[str, Union[StrictStr, AnyUrl]]:
@model_validator(mode="after")
def validate_spdx_identifier(self) -> "License":
"""Validate the SPDX license identifier."""
spdx_id = values["identifier"]
spdx_id = self.identifier
try:
get_spdx_licensing().parse(spdx_id, validate=True)
if spdx_id not in [None, ""] and values.get("text_url") is None:
values["text_url"] = cls._get_license_text_url(spdx_id) # type: ignore
if spdx_id not in [None, ""] and self.text_url is None:
text_url = self._get_license_text_url(spdx_id)
# bypass the validated `__setattr__` (which would re-run this
# "after" validator and recurse indefinitely); coerce to
# `AnyUrl` ourselves since validation is being skipped
object.__setattr__(
self,
"text_url",
AnyUrl(text_url) if text_url is not None else None,
)
except ExpressionError as exc:
if spdx_id.lower() not in ["proprietary", "unlicensed", "unknown"]:
if spdx_id is None or spdx_id.lower() not in [
"proprietary",
"unlicensed",
"unknown",
]:
raise ValueError(
"Expected a valid SPDX license identifier "
f"(https://spdx.org/licenses/). Got {spdx_id} instead.",
) from exc
return values
return self

@staticmethod
def _get_license_text_url(identifier: Optional[str]) -> Optional[str]:
Expand Down Expand Up @@ -154,9 +163,10 @@ class Citation(
description="The citation content in BibTeX format.",
)

@validator("content")
@field_validator("content")
@classmethod
def parse_content(
cls: "Citation", # noqa: N805
cls: "type[Citation]", # noqa: N805
value: StrictStr,
) -> StrictStr:
"""Parse the citation content."""
Expand Down Expand Up @@ -226,12 +236,10 @@ class SensitiveData(BaseModelCardField, composable_with=["Dataset"], list_factor
"aggregated. Please describe any such fields here."
),
default_factory=list,
unique_items=True,
)
sensitive_data_used: Optional[List[StrictStr]] = Field(
description="A list of sensitive data used in the deployed model.",
default_factory=list,
unique_items=True,
)
justification: Optional[StrictStr] = Field(
None,
Expand All @@ -253,23 +261,19 @@ class Dataset(BaseModelCardField, composable_with=["Datasets"], list_factory=Tru
citations: Optional[List[Citation]] = Field(
description="How should the dataset be cited?",
default_factory=list,
unique_items=True,
)
references: Optional[List[Reference]] = Field(
description="Provide any additional links to resources the reader may need.",
default_factory=list,
unique_items=True,
)
licenses: Optional[List[License]] = Field(
description="The license information for the dataset.",
default_factory=list,
unique_items=True,
)
version: Optional[Version] = Field(None, description="The version of the dataset.")
features: Optional[List[StrictStr]] = Field(
description="A list of features in the dataset.",
default_factory=list,
unique_items=True,
)
graphics: Optional[GraphicsCollection] = Field(
None,
Expand Down Expand Up @@ -419,9 +423,10 @@ class UseCase(
),
)

@validator("kind")
@field_validator("kind")
@classmethod
def kind_must_be_valid(
cls: "UseCase", # noqa: N805
cls: "type[UseCase]", # noqa: N805
value: str,
) -> str:
"""Validate the use case kind."""
Expand Down Expand Up @@ -595,7 +600,7 @@ class MetricCard(
)

history: List[StrictFloat] = Field(
None,
default_factory=list,
description="History of the metric over time.",
)

Expand Down
Loading
Loading