Source code for ibek.ioc
"""
Classes for generating an IocInstance derived class from a set of
support module YAML files
"""
from __future__ import annotations
import ast
import builtins
from enum import Enum
from typing import Any, Dict, List, Sequence
from pydantic import (
Field,
model_validator,
)
from pydantic.fields import FieldInfo
from .entity_model import EntityModel
from .globals import BaseSettings
from .utils import UTILS
# a global dict of all entity instances indexed by their ID
id_to_entity: Dict[str, Entity] = {}
def get_entity_by_id(id: str) -> Entity:
try:
return id_to_entity[id]
except KeyError:
raise ValueError(f"object {id} not found in {list(id_to_entity)}")
[docs]
def clear_entity_model_ids():
"""Resets the global id_to_entity dict"""
id_to_entity.clear()
[docs]
class EnumVal(Enum):
"""
An enum that is printed as its name only
"""
def __str__(self):
return self.name
[docs]
class Entity(BaseSettings):
"""
A baseclass for all generated Entity classes.
"""
type: str = Field(description="The type of this entity")
entity_enabled: bool = Field(
description="enable or disable this entity instance", default=True
)
_model: EntityModel
def _process_field(self: Entity, name: str, value: Any, typ: str):
"""
Process an Entity field - doing jinja rendering, type coercion and
object id storage/lookup as required.
"""
if isinstance(value, str):
# Jinja expansion always performed on string fields
value = UTILS.render(self, value)
if typ in ["list", "int", "float", "bool"]:
# coerce the rendered parameter to its intended type
try:
cast_type = getattr(builtins, typ)
value = cast_type(ast.literal_eval(value))
except:
print(f"ERROR: decoding field '{name}', value '{value}' as {typ}")
raise
if typ == "object":
# look up the actual object by it's id
if isinstance(value, str):
value = get_entity_by_id(value)
# If this field is not pre-existing, add it into the model instance.
# This is how pre/post_defines are added.
if name not in self.model_fields:
self.model_fields[name] = FieldInfo(annotation=str, default=value)
# update the model instance attribute with the rendered value
setattr(self, name, value)
if typ == "id":
# add this entity to the global id index
if value in id_to_entity:
raise ValueError(f"Duplicate id {value} in {list(id_to_entity)}")
id_to_entity[value] = self
[docs]
@model_validator(mode="after")
def add_ibek_attributes(self):
"""
Whole Entity model validation
Do jinja rendering of pre_defines/ parameters / post_defines
in the correct order.
Also adds pre_defines and post_defines to the model instance, making
them available for the phase 2 (final) jinja rendering performed in
ibek.runtime_cmds.generate().
"""
if self._model.pre_defines:
for name, define in self._model.pre_defines.items():
self._process_field(name, define.value, define.type)
if self._model.parameters:
for name, parameter in self._model.parameters.items():
self._process_field(name, getattr(self, name), parameter.type)
if self._model.post_defines:
for name, define in self._model.post_defines.items():
self._process_field(name, define.value, define.type)
return self
def __str__(self):
"""
When a jinja template refers to an object by itself e.g.
# this is the startup entry for {{ my_entity }}
Jinja will attempt to render the object as a string and this
method will be called.
The behaviour is to print the ID of the object. Thus we look up
which of our object's fields is the ID field and return the
value of that field.
"""
id_name = self._model._get_id_arg()
if id_name:
return getattr(self, id_name)
else:
raise ValueError(f"Entity {self} has no id field")
def __repr__(self):
return str(self)
[docs]
class IOC(BaseSettings):
"""
Used to load an IOC instance entities yaml file into a Pydantic Model.
"""
ioc_name: str = Field(description="Name of IOC instance")
description: str = Field(description="Description of what the IOC does")
entities: List[Entity]
shared: Sequence[Any] = Field(
description="A place to create any anchors required for repeating YAML",
default=(),
)