utils

Attributes

Config module-attribute

Config = TypeVar('Config', bound=BaseModel)

Classes

ConfigError

Bases: Exception

Overrideable

Bases: YamlModel, GenericModel, Generic[Config]

Attributes

overrides class-attribute instance-attribute

overrides: Dict[str, Any] = Field(
    default_factory=dict,
    description="Overrides to apply to the config file referenced by `path`.\nOverrides are defined in `key`: `value` pairs, where the `key` is a pointer to the object in the config file to override and the `value` is what should replace it.\nThe format of the keys is a cross between path-like structures and a python dictionary. For example, to change the 'location_id' property on the python object `obj = {'attrs': {'location_id': 'abc'}, 'data_vars': {...}}` to 'sgp' you would write `/attrs/location_id: 'sgp'`.\nOverrides are implemented using https://python-json-pointer.readthedocs.io/en/latest/tutorial.html",
)

path class-attribute instance-attribute

path: FilePath = Field(
    description="Path to the configuration file to borrow configurations from.\nNote that this path is relative to the project root, so you should include any paths in between the project root and your config file.\nE.g., `pipelines/lidar/config/dataset.yaml`"
)

ParameterizedConfigClass

Bases: BaseModel

Attributes

classname class-attribute instance-attribute

classname: StrictStr = Field(
    description="The import path to the Python class that should be used, e.g., if your import statement looks like `from foo.bar import Baz`, then your classname would be `foo.bar.Baz`."
)

parameters class-attribute instance-attribute

parameters: Dict[str, Any] = Field(
    {},
    description="Optional dictionary that will be passed to the Python class specified by 'classname' when it is instantiated. If the object is a tsdat class, then the parameters will typically be made accessible under the `params` property on an instance of the class. See the documentation for individual classes for more information.",
)

Functions

classname_looks_like_a_module classmethod

classname_looks_like_a_module(v: StrictStr) -> StrictStr

Source code in tsdat/config/utils.py

@validator("classname")
@classmethod
def classname_looks_like_a_module(cls, v: StrictStr) -> StrictStr:
    if "." not in v or not v.replace(".", "").replace("_", "").isalnum():
        raise ValueError(f"Classname '{v}' is not a valid classname.")
    return v

instantiate

instantiate() -> Any

---

Instantiates and returns the class specified by the 'classname' parameter.

Returns:

Name	Type	Description
Any	Any	An instance of the specified class.

---

Source code in tsdat/config/utils.py

def instantiate(self) -> Any:
    """------------------------------------------------------------------------------------
    Instantiates and returns the class specified by the 'classname' parameter.

    Returns:
        Any: An instance of the specified class.

    ------------------------------------------------------------------------------------
    """
    params = {field: getattr(self, field) for field in self.__fields_set__}
    _cls = import_string(params.pop("classname"))
    return _cls(**params)

YamlModel

Bases: BaseModel

Functions

from_yaml classmethod

from_yaml(
    filepath: Path,
    overrides: Optional[Dict[str, Any]] = None,
)

---

Creates a python configuration object from a yaml file.

Parameters:

Name	Type	Description	Default
filepath	Path	The path to the yaml file	required
overrides	Optional[Dict[str, Any]]	Overrides to apply to the yaml before instantiating the YamlModel object. Defaults to None.	None

Returns:

Name	Type	Description
YamlModel		A YamlModel subclass

---

Source code in tsdat/config/utils.py

@classmethod
def from_yaml(cls, filepath: Path, overrides: Optional[Dict[str, Any]] = None):
    """------------------------------------------------------------------------------------
    Creates a python configuration object from a yaml file.

    Args:
        filepath (Path): The path to the yaml file
        overrides (Optional[Dict[str, Any]], optional): Overrides to apply to the
            yaml before instantiating the YamlModel object. Defaults to None.

    Returns:
        YamlModel: A YamlModel subclass

    ------------------------------------------------------------------------------------
    """
    config = read_yaml(filepath)
    if overrides:
        for pointer, new_value in overrides.items():
            set_pointer(config, pointer, new_value)
    try:
        return cls(**config)
    except (ValidationError, Exception) as e:
        raise ConfigError(
            f"Error encountered while instantiating {filepath}"
        ) from e

generate_schema classmethod

generate_schema(output_file: Path)

---

Generates JSON schema from the model fields and type annotations.

Parameters:

Name	Type	Description	Default
output_file	Path	The path to store the JSON schema.	required

---

Source code in tsdat/config/utils.py

@classmethod
def generate_schema(cls, output_file: Path):
    """------------------------------------------------------------------------------------
    Generates JSON schema from the model fields and type annotations.

    Args:
        output_file (Path): The path to store the JSON schema.

    ------------------------------------------------------------------------------------
    """
    output_file.write_text(cls.schema_json(indent=4))

Functions

find_duplicates

find_duplicates(
    entries: Sequence[_NamedClass],
) -> List[str]

Source code in tsdat/config/utils.py

def find_duplicates(entries: Sequence[_NamedClass]) -> List[str]:
    duplicates: List[str] = []
    seen: Set[str] = set()
    for entry in entries:
        if entry.name in seen:
            duplicates.append(entry.name)
        else:
            seen.add(entry.name)
    return duplicates

get_code_version

get_code_version() -> str

Source code in tsdat/config/utils.py

def get_code_version() -> str:
    version = "N/A"
    try:
        version = os.environ["CODE_VERSION"]
    except KeyError:
        try:
            version = Version.from_git().serialize(dirty=True, style=Style.SemVer)
        except RuntimeError:
            logger.warning(
                "Could not get code_version from either the 'CODE_VERSION' environment"
                " variable nor from git history. The 'code_version' global attribute"
                " will be set to 'N/A'.",
            )
    return version

matches_overrideable_schema

matches_overrideable_schema(model_dict: Dict[str, Any])

Source code in tsdat/config/utils.py

def matches_overrideable_schema(model_dict: Dict[str, Any]):
    return "path" in model_dict

read_yaml

read_yaml(filepath: Path) -> Dict[Any, Any]

Source code in tsdat/config/utils.py

def read_yaml(filepath: Path) -> Dict[Any, Any]:
    return list(yaml.safe_load_all(filepath.read_text(encoding="UTF-8")))[0]

recursive_instantiate

recursive_instantiate(model: Any) -> Any

---

Instantiates all ParametrizedClass components and subcomponents of a given model.

Recursively calls model.instantiate() on all ParameterizedConfigClass instances under the the model, resulting in a new model which follows the same general structure as the given model, but possibly containing totally different properties and methods.

Note that this method does a depth-first traversal of the model tree to to instantiate leaf nodes first. Traversing breadth-first would result in new pydantic models attempting to call the init method of child models, which is not valid because the child models are ParameterizedConfigClass instances. Traversing depth-first allows us to first transform child models into the appropriate type using the classname of the ParameterizedConfigClass.

This method is primarily used to instantiate a Pipeline subclass and all of its properties from a yaml pipeline config file, but it can be applied to any other pydantic model.

Parameters:

Name	Type	Description	Default
model	Any	The object to recursively instantiate.	required

Returns:

Name	Type	Description
Any	Any	The recursively-instantiated object.

---

Source code in tsdat/config/utils.py

def recursive_instantiate(model: Any) -> Any:
    """---------------------------------------------------------------------------------
    Instantiates all ParametrizedClass components and subcomponents of a given model.

    Recursively calls model.instantiate() on all ParameterizedConfigClass instances under
    the the model, resulting in a new model which follows the same general structure as
    the given model, but possibly containing totally different properties and methods.

    Note that this method does a depth-first traversal of the model tree to to
    instantiate leaf nodes first. Traversing breadth-first would result in new pydantic
    models attempting to call the __init__ method of child models, which is not valid
    because the child models are ParameterizedConfigClass instances. Traversing
    depth-first allows us to first transform child models into the appropriate type
    using the classname of the ParameterizedConfigClass.

    This method is primarily used to instantiate a Pipeline subclass and all of its
    properties from a yaml pipeline config file, but it can be applied to any other
    pydantic model.

    Args:
        model (Any): The object to recursively instantiate.

    Returns:
        Any: The recursively-instantiated object.

    ---------------------------------------------------------------------------------"""
    # Case: ParameterizedConfigClass. Want to instantiate any sub-models then return the
    # class with all sub-models recursively instantiated, then statically instantiate
    # the model. Note: the model is instantiated last so that sub-models are only
    # processed once.
    if isinstance(model, ParameterizedConfigClass):
        fields = model.__fields_set__ - {"classname"}  # No point checking classname
        for field in fields:
            setattr(model, field, recursive_instantiate(getattr(model, field)))
        model = model.instantiate()

    # Case: BaseModel. Want to instantiate any sub-models then return the model itself.
    elif isinstance(model, BaseModel):
        fields = model.__fields_set__
        if "classname" in fields:
            raise ValueError(
                f"Model '{model.__repr_name__()}' provides a 'classname' but does not"
                " extend ParametrizedConfigClass."
            )
        for field in fields:
            setattr(model, field, recursive_instantiate(getattr(model, field)))

    # Case: List. Want to iterate through and recursively instantiate all sub-models in
    # the list, then return everything as a list.
    elif isinstance(model, List):
        model = [recursive_instantiate(m) for m in cast(List[Any], model)]

    # Case Dict. Want to iterate through and recursively instantiate all sub-models in
    # the Dict's values, then return everything as a Dict, unless the dict is meant to
    # be turned into a parameterized class, in which case we instantiate it as the
    # intended object
    elif isinstance(model, Dict):
        model = {
            k: recursive_instantiate(v) for k, v in cast(Dict[str, Any], model).items()
        }
        if "classname" in model:
            classname: str = model.pop("classname")  # type: ignore
            _cls = import_string(classname)
            return _cls(**model)

    return model