"""The Template node implementation"""
# pylint: disable= invalid-name, too-many-instance-attributes
import logging
import contextvars as cv
from dataclasses import dataclass
import logging
import attr
from .sequence import Sequence
from .. import ob
from .utils import seq_factory
user_logger = logging.getLogger("seq.user")
logger = logging.getLogger(__name__)
@dataclass
class TemplateParam:
"""Holds the name and value of some Template Parameter"""
name: str
value: None
[docs]@attr.s
class Template(Sequence):
"""The Template node
A :class:`Template` is just a :class:`Sequence` node with access to a set
of variables as defined in an :class:`ObservingBlock`.
The :class:`ObservingBlock` takes care of instantiating the :class:`Template`
objects as needed by the Observation Block.
============ =======================================
Context Variables
-----------------------------------------------------
Name Desc
============ =======================================
current_tpl The current Template being executed
============ =======================================
.. warning:: Objects of type :class:`Template` should not be directly constructed.
Leave that for the :class:`ObservingBlock`.
"""
tpl_name = attr.ib(default=None)
# To get default param values from file.
params = attr.ib(default=attr.Factory(list), repr=False)
content = attr.ib(default=attr.Factory(dict), repr=False)
mod = attr.ib(default=None, repr=False, init=False)
_validator = attr.ib(default=attr.Factory(dict), repr=False, init=False)
# context variables
P = cv.ContextVar("P", default={})
current_tpl = cv.ContextVar("current_tpl", default=None)
def __attrs_post_init__(self):
"""
Assigns node's name and id.
"""
for el in self.params:
self._parameters[el["name"]] = el
super().__attrs_post_init__()
[docs] async def execute(self, resume=False, propagate=False):
"""Executes node -- this just creates the asyncio task"""
Template.P.set(self._parameters)
logger.info("execute templates: %s", self.parameters)
return await super().execute(resume, propagate=True)
[docs] def make_sequence(self, parent_tpl=None):
"""Builds this sequence execution graph.
Joins the Sequence's nodes together.
"""
super().make_sequence(parent_tpl=self)
[docs] @staticmethod
def from_dict(d):
"""
Creates a Template from keywords in dictionary.
"""
t = Template(
tpl_name=d["templateName"],
params=d.get("parameters", {}),
content=d,
)
user_logger.info("Loading Template module: %s", t.tpl_name)
t.mod = ob.ModuleLoader.from_mod(t.tpl_name)
return t
# pylint: disable=arguments-differ
[docs] @staticmethod
def create(d, *args, **kw):
"""Creates a :class:`Template` node
Args:
*args: Variable length list of nodes or coroutines that composes the sequence.
Keyword Args:
id: Node id.
name: node name.
"""
a = Template.from_dict(d)
# get module's and ctor documentation
ctor_docs = getattr(a.mod.ctor, "__doc__", None) or ""
mod_docs = getattr(a.mod.module, "__doc__", None) or ""
a.description = ctor_docs.strip().split("\n")[0] or "No description"
a.name = mod_docs.strip().split("\n")[0] or "Template"
# creates the actual sequence
a.seq = [seq_factory(a.mod, *args, **kw)]
return a