Module fontai.config.pipeline
Expand source code
import typing as t
import logging
import strictyaml as yml
from fontai.config.core import BaseConfigHandler, BasePipelineTransformConfig
import fontai.runners.stages as stages
logger = logging.getLogger(__name__)
class Config(BasePipelineTransformConfig):
"""Configuration object for Pipeline stage instances
Attributes:
stages (t.List[type]): List with types of pipeline stages
configs: (t.List[BasePipelineTransformConfig]): List with parsed configuration objects for each stage
"""
stages: t.List[type]
configs: t.List[BasePipelineTransformConfig]
fit_stage: t.List[bool]
class ConfigHandler(BaseConfigHandler):
STAGE_TYPES = {
"ingestion": stages.Ingestion,
"preprocessing": stages.Preprocessing,
"scoring": stages.Scoring
}
@classmethod
def get_config_schema(self):
"""
YAML configuration schema:
stages: List of subYAMLs with keys (type, config), where type is one of 'ingestion', 'preprocessing' or 'scoring', and config is a YAML matching the config schema for the corresponding type
"""
schema = yml.Map({
"stages" : yml.Seq(yml.Map({
"type": yml.Str(),
yml.Optional("fit", default=False): yml.Bool(),
"config": yml.Any() #configs will be matched to a schema later
})
)
})
return schema
def instantiate_config(self, config: yml.YAML) -> Config:
"""
Processes a YAML instance to produce an Config instance.
Args:
config (yml.YAML): YAML object from the strictyaml library
Returns:
Config: Instantiated configuration
"""
stages = []
configs = []
fit_stage = []
for entry in config.get("stages"):
stages.append(self.STAGE_TYPES[entry.get("type").text])
configs.append(entry.get("config"))
fit_stage.append(entry.get("fit"))
configs = [stage.parse_config_file(cfg) for stage, cfg in zip(stages, configs)]
#configs = [getattr(stages, stage.get("class").text).parse_config_file(stage.get("yaml_config_path").text) for stage in config.get("stages")]
return Config(
stages = stages,
configs = configs,
fit_stage = fit,
yaml = config)
Classes
class Config (**data: Any)
-
Configuration object for Pipeline stage instances
Attributes
stages
:t.List[type]
- List with types of pipeline stages
configs
- (t.List[BasePipelineTransformConfig]): List with parsed configuration objects for each stage
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class Config(BasePipelineTransformConfig): """Configuration object for Pipeline stage instances Attributes: stages (t.List[type]): List with types of pipeline stages configs: (t.List[BasePipelineTransformConfig]): List with parsed configuration objects for each stage """ stages: t.List[type] configs: t.List[BasePipelineTransformConfig] fit_stage: t.List[bool]
Ancestors
- BasePipelineTransformConfig
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var configs : List[BasePipelineTransformConfig]
var fit_stage : List[bool]
var stages : List[type]
class ConfigHandler
-
Interface for creating execution configuration objects for ML pipeline stages
Attributes
IO_CONFIG_SCHEMA
:yml.validators.Validator
- schema for I/O parameters
yaml_to_obj
:SimpleClassInstantiator
- Helper class to instantiate some Python objects from a YAML definition
Expand source code
class ConfigHandler(BaseConfigHandler): STAGE_TYPES = { "ingestion": stages.Ingestion, "preprocessing": stages.Preprocessing, "scoring": stages.Scoring } @classmethod def get_config_schema(self): """ YAML configuration schema: stages: List of subYAMLs with keys (type, config), where type is one of 'ingestion', 'preprocessing' or 'scoring', and config is a YAML matching the config schema for the corresponding type """ schema = yml.Map({ "stages" : yml.Seq(yml.Map({ "type": yml.Str(), yml.Optional("fit", default=False): yml.Bool(), "config": yml.Any() #configs will be matched to a schema later }) ) }) return schema def instantiate_config(self, config: yml.YAML) -> Config: """ Processes a YAML instance to produce an Config instance. Args: config (yml.YAML): YAML object from the strictyaml library Returns: Config: Instantiated configuration """ stages = [] configs = [] fit_stage = [] for entry in config.get("stages"): stages.append(self.STAGE_TYPES[entry.get("type").text]) configs.append(entry.get("config")) fit_stage.append(entry.get("fit")) configs = [stage.parse_config_file(cfg) for stage, cfg in zip(stages, configs)] #configs = [getattr(stages, stage.get("class").text).parse_config_file(stage.get("yaml_config_path").text) for stage in config.get("stages")] return Config( stages = stages, configs = configs, fit_stage = fit, yaml = config)
Ancestors
- BaseConfigHandler
- abc.ABC
Class variables
var STAGE_TYPES
Static methods
def get_config_schema()
-
YAML configuration schema:
stages: List of subYAMLs with keys (type, config), where type is one of 'ingestion', 'preprocessing' or 'scoring', and config is a YAML matching the config schema for the corresponding type
Expand source code
@classmethod def get_config_schema(self): """ YAML configuration schema: stages: List of subYAMLs with keys (type, config), where type is one of 'ingestion', 'preprocessing' or 'scoring', and config is a YAML matching the config schema for the corresponding type """ schema = yml.Map({ "stages" : yml.Seq(yml.Map({ "type": yml.Str(), yml.Optional("fit", default=False): yml.Bool(), "config": yml.Any() #configs will be matched to a schema later }) ) }) return schema
Methods
def instantiate_config(self, config: strictyaml.representation.YAML) ‑> Config
-
Processes a YAML instance to produce an Config instance.
Args
config
:yml.YAML
- YAML object from the strictyaml library
Returns
Config
- Instantiated configuration
Expand source code
def instantiate_config(self, config: yml.YAML) -> Config: """ Processes a YAML instance to produce an Config instance. Args: config (yml.YAML): YAML object from the strictyaml library Returns: Config: Instantiated configuration """ stages = [] configs = [] fit_stage = [] for entry in config.get("stages"): stages.append(self.STAGE_TYPES[entry.get("type").text]) configs.append(entry.get("config")) fit_stage.append(entry.get("fit")) configs = [stage.parse_config_file(cfg) for stage, cfg in zip(stages, configs)] #configs = [getattr(stages, stage.get("class").text).parse_config_file(stage.get("yaml_config_path").text) for stage in config.get("stages")] return Config( stages = stages, configs = configs, fit_stage = fit, yaml = config)
Inherited members