Source code for ax.storage.sqa_store.sqa_config

#!/usr/bin/env python3
# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

# pyre-strict

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, Optional, Type, Union

from ax.analysis.analysis_report import AnalysisReport
from ax.analysis.base_analysis import BaseAnalysis

from ax.core.arm import Arm
from ax.core.batch_trial import AbandonedArm
from ax.core.data import Data
from ax.core.experiment import Experiment
from ax.core.generator_run import GeneratorRun, GeneratorRunType
from ax.core.metric import Metric
from ax.core.parameter import Parameter
from ax.core.parameter_constraint import ParameterConstraint
from ax.core.runner import Runner
from ax.core.trial import Trial
from ax.modelbridge.generation_strategy import GenerationStrategy
from ax.storage.json_store.registry import (
    CORE_CLASS_DECODER_REGISTRY,
    CORE_CLASS_ENCODER_REGISTRY,
    CORE_DECODER_REGISTRY,
    CORE_ENCODER_REGISTRY,
    TDecoderRegistry,
)
from ax.storage.metric_registry import CORE_METRIC_REGISTRY
from ax.storage.runner_registry import CORE_RUNNER_REGISTRY
from ax.storage.sqa_store.db import SQABase
from ax.storage.sqa_store.sqa_classes import (
    SQAAbandonedArm,
    SQAAnalysis,
    SQAAnalysisReport,
    SQAArm,
    SQAData,
    SQAExperiment,
    SQAGenerationStrategy,
    SQAGeneratorRun,
    SQAMetric,
    SQAParameter,
    SQAParameterConstraint,
    SQARunner,
    SQATrial,
)
from ax.utils.common.base import Base


[docs]@dataclass class SQAConfig: """Metadata needed to save and load an experiment to SQLAlchemy. Attributes: class_to_sqa_class: Mapping of user-facing class to SQLAlchemy class that it will be encoded to. This allows overwriting of the default classes to provide custom save functionality. experiment_type_enum: Enum containing valid Experiment types. generator_run_type_enum: Enum containing valid Generator Run types. json_encoder_registry: Mapping from user-facing types to their json serialization function. """ def _default_class_to_sqa_class(self=None) -> Dict[Type[Base], Type[SQABase]]: # pyre-ignore [7] return { AbandonedArm: SQAAbandonedArm, Arm: SQAArm, Data: SQAData, Experiment: SQAExperiment, GenerationStrategy: SQAGenerationStrategy, GeneratorRun: SQAGeneratorRun, Parameter: SQAParameter, ParameterConstraint: SQAParameterConstraint, Metric: SQAMetric, Runner: SQARunner, Trial: SQATrial, BaseAnalysis: SQAAnalysis, AnalysisReport: SQAAnalysisReport, } class_to_sqa_class: Dict[Type[Base], Type[SQABase]] = field( default_factory=_default_class_to_sqa_class ) experiment_type_enum: Optional[Union[Enum, Type[Enum]]] = None generator_run_type_enum: Optional[Union[Enum, Type[Enum]]] = GeneratorRunType # pyre-fixme[4]: Attribute annotation cannot contain `Any`. # pyre-fixme[24]: Generic type `type` expects 1 type parameter, use # `typing.Type` to avoid runtime subscripting errors. json_encoder_registry: Dict[Type, Callable[[Any], Dict[str, Any]]] = field( default_factory=lambda: CORE_ENCODER_REGISTRY ) # pyre-fixme[4]: Attribute annotation cannot contain `Any`. # pyre-fixme[24]: Generic type `type` expects 1 type parameter, use # `typing.Type` to avoid runtime subscripting errors. json_class_encoder_registry: Dict[Type, Callable[[Any], Dict[str, Any]]] = field( default_factory=lambda: CORE_CLASS_ENCODER_REGISTRY ) json_decoder_registry: TDecoderRegistry = field( default_factory=lambda: CORE_DECODER_REGISTRY ) # pyre-fixme[4]: Attribute annotation cannot contain `Any`. json_class_decoder_registry: Dict[str, Callable[[Dict[str, Any]], Any]] = field( default_factory=lambda: CORE_CLASS_DECODER_REGISTRY ) metric_registry: Dict[Type[Metric], int] = field( default_factory=lambda: CORE_METRIC_REGISTRY ) runner_registry: Dict[Type[Runner], int] = field( default_factory=lambda: CORE_RUNNER_REGISTRY ) @property def reverse_metric_registry(self) -> Dict[int, Type[Metric]]: return {v: k for k, v in self.metric_registry.items()} @property def reverse_runner_registry(self) -> Dict[int, Type[Runner]]: return {v: k for k, v in self.runner_registry.items()}