"""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
user_logger = logging.getLogger("seq.user")
# value_converter = {"string": str, "number": float, "integer": int}
@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)
_parameters = attr.ib(default=attr.Factory(dict), 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.
"""
if self.name is None:
self.name = "Template"
super().__attrs_post_init__()
for el in self.params:
self._parameters[el["name"]] = el
@property
def parameters(self):
"""Return parameters"""
return self._parameters
[docs] def par(self, k):
"""Get a parameter value"""
return self._parameters[k]["value"]
[docs] def set(self, p):
"""Sets the value of a parameter"""
if not p.name in self._parameters:
raise KeyError(p.name)
self._parameters[p.name]["value"] = p.value
[docs] def set_params(self, params):
""""Update parameters from list"""
#logger = logging.getLogger(__name__)
for k,v in params.items():
self._parameters[k] = v
[docs] async def execute(self, resume=False):
"""Executes node -- this just creates the asyncio task"""
Template.P.set(self._parameters)
Template.current_tpl.set(self)
return await super().execute(resume)
[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)
a.seq = [ob.OB.create_sequence(a.mod, *args, **kw)]
return a