Config

def to_concrete(
    self,
    model_params: type[ModelParamsLikeT],
    *,
    pydantic_config: ConfigDict = ConfigDict(extra="forbid"),
) -> ModelParamsLikeT:
    """Validate against a real set of model parameters and convert to it.

    :raises pydantic.ValidationError: if extra fields are present in the model parameters
        that doesn't exist in the concrete model parameters.
    """
    # input_type and output_type are inconfigurable anyway
    # TODO: use lru cache to avoid recreating the TypeAdapter in a hot loop but dict isn't hashable
    ta = TypeAdapter(
        type(
            f"{model_params.__name__}Validator",
            (model_params,),
            {"__pydantic_config__": pydantic_config},
        )  # needed for https://docs.pydantic.dev/latest/errors/usage_errors/#type-adapter-config-unused
    )  # type: ignore
    # types defined within `TYPE_CHECKING` blocks will be forward references, so we need rebuild
    ta.rebuild(_types_namespace={"TorchDtype": TorchDtype, "t": t})
    model_params_concrete: ModelParamsLikeT = ta.validate_python(self.model_dump())
    return model_params_concrete

@model_validator(mode="after")
def check_derived_stems(self) -> Self:
    if self.derived_stems is None:
        return self

    base_output_stems = tuple(self.inference.requested_stems or self.model.output_stem_names)
    existing_stem_names: list[StemName] = [*_INPUT_STEM_NAMES, *base_output_stems]
    for derived_stem_name, definition in self.derived_stems.items():
        if derived_stem_name in existing_stem_names:
            raise PydanticCustomError(
                "derived_stem_name_conflict",
                "Derived stem `{derived_stem_name}` must not conflict with existing stem names: `{existing_stem_names}`",
                {
                    "derived_stem_name": derived_stem_name,
                    "existing_stem_names": existing_stem_names,
                },
            )
        required_stems: tuple[StemName, ...] = tuple()
        if isinstance(definition, SubtractConfig):
            required_stems = (definition.stem_name, definition.by_stem_name)
        elif isinstance(definition, SumConfig):
            required_stems = definition.stem_names
        for stem_name in required_stems:
            if stem_name not in existing_stem_names:
                raise PydanticCustomError(
                    "invalid_derived_stem",
                    "Derived stem `{derived_stem_name}` requires stem `{stem_name}` but is not found in `{existing_stem_names}`",
                    {
                        "derived_stem_name": derived_stem_name,
                        "stem_name": stem_name,
                        "existing_stem_names": existing_stem_names,
                    },
                )
        existing_stem_names.append(derived_stem_name)
    return self

@model_validator(mode="after")
def check_requested_stems(self) -> Self:
    if self.inference.requested_stems is None:
        return self

    model_stem_names = set(self.model.output_stem_names)
    for stem_name in self.inference.requested_stems:
        if stem_name in model_stem_names:
            continue
        raise PydanticCustomError(
            "invalid_target_stem",
            "Target stem `{stem_name}` is not found in model output stems: `{model_stem_names}`",
            {
                "stem_name": stem_name,
                "model_stem_names": self.model.output_stem_names,
            },
        )

    return self

def validate_inference_contract(self, model_params: ModelParamsLike) -> t.InferenceArchetype:
    archetype = model_params.inference_archetype
    input_type = model_params.input_type
    output_type = model_params.output_type

    if archetype == "standard_end_to_end":
        if self.waveform_chunking is None:
            raise ValueError("`waveform_chunking` is required for standard_end_to_end pipeline")
        if output_type != "waveform":
            raise ValueError(
                f"standard_end_to_end expects waveform model output, got {output_type}"
            )
    elif archetype == "frequency_masking":
        if self.transform is None or self.transform.kind != "stft":
            raise ValueError(
                '`transform` with `kind="stft"` is required for frequency_masking pipeline'
            )
        if self.waveform_chunking is None:
            raise ValueError("`waveform_chunking` is required for frequency_masking pipeline")
        if input_type not in {"spectrogram", "waveform_and_spectrogram"}:
            raise ValueError(
                f"frequency_masking expects spectrogram-like model input, got {input_type}"
            )
        if output_type not in {"spectrogram_mask", "spectrogram"}:
            raise ValueError(
                f"frequency_masking expects spectrogram-like model output, got {output_type}"
            )
    elif archetype == "sequence_labeling":
        if self.sequence_chunking is None:
            raise ValueError("`sequence_chunking` is required for sequence_labeling pipeline")
        if input_type == "spectrogram" and self.transform is None:
            raise ValueError("spectrogram sequence models require `transform` to be configured")
        if input_type not in {"spectrogram", "waveform"}:
            raise ValueError(
                f"sequence_labeling expects spectrogram or waveform model input, got {input_type}"
            )
        if output_type not in {"logits", "multi_stream"}:
            raise ValueError(
                f"sequence_labeling expects logits or multi_stream model output, got {output_type}"
            )
        if input_type == "spectrogram" and not isinstance(
            model_params, HasSequenceWindowFrames
        ):
            raise ValueError(
                "spectrogram sequence models must declare `sequence_window_frames` explicitly"
            )
    else:
        assert_never(archetype)

    if output_type in {"logits", "multi_stream"} and self.output.file_format != "npy":
        raise ValueError(
            f"sequence models require output.file_format='npy', got {self.output.file_format!r}"
        )
    if output_type not in {"logits", "multi_stream"} and self.output.file_format == "npy":
        raise ValueError("waveform/spectrogram models cannot use output.file_format='npy'")

    return archetype

@classmethod
def from_file(
    cls,
    path: t.StrPath | t.BytesPath,
    *,
    overrides: ConfigOverrides = (),
) -> Config:
    """Load config JSON from disk, optionally applying CLI-style overrides."""
    config_dict = load_config_dict(path)
    if overrides:
        apply_config_overrides(config_dict, overrides)
    return cls.model_validate(config_dict)

@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    return core_schema.no_info_after_validator_function(cls, handler(dict[t.Identifier, Model]))

@classmethod
def from_file(cls, path: t.StrPath | t.BytesPath) -> Registry:
    with open(path, "r") as f:
        data = f.read()
    ta = TypeAdapter(cls)
    return ta.validate_json(data)