"""Observing Block class.
Represents a Sequence from an OB file (json).
"""
# pylint: disable=invalid-name, too-many-instance-attributes, logging-format-interpolation
import json
import logging
import attr
from .sequence import Sequence
from .template import Template
from ..counter import Counter
value_converter = {"string": str, "number": float, "integer": int}
logger = logging.getLogger(__name__)
user_logger = logging.getLogger("seq.user")
[docs]@attr.s
class ObservingBlock(Sequence):
"""
Represents a Sequence from an OB file (json).
The JSON file contains a lot of data, however the Sequencer only cares
about the **templates** section::
"templates": [
{
"templateName": "seq.samples.a",
"type": "string"
},
{
"templateName": "seq.samples.tpa",
"type": "string",
"parameters": [
{
"name": "par_b",
"type": "integer",
"value": 0
},
{
"name": "par_c",
"type": "number",
"value": 77
}
]
}
],
Inside the **templates** section there is list of *templateName* objects.
The *templateName* defines a Python module that can be directly
imported (it is reachable from PYTHONPATH).
Each template might contain a list of *parameters* that can be accessed
from corresponding python code that implements the template.
"""
fname = attr.ib(default=None)
content = attr.ib(default=attr.Factory(dict), repr=False, init=False)
template_parameters = {}
current_tpl_params = attr.ib(default=attr.Factory(dict), init=False)
ctor = attr.ib(default=None, init=False)
descr = attr.ib(default=attr.Factory(dict), init=False)
def __attrs_post_init__(self):
"""
Assigns node's name and id.
"""
if self.name is None:
self.name = "OB"
super().__attrs_post_init__()
self.load()
@property
def parameters(self):
"""Returns a dictionary with template's parameters
The returned dictionary enumerates the OB's templates as keys and the values are
Template's parameters
"""
return {k: t.parameters for k, t in enumerate(self.seq)}
[docs] def load(self):
"""
Loads json file
"""
user_logger.info("Loading OB: %s", self.fname)
assert self.fname
with open(self.fname) as fd:
self.content = json.load(fd)
self.descr = self.content["obsDescription"]
self.name = self.content["obsDescription"]["name"]
def _set_parameters_from_dict(self, d):
for k, tvars in d.items():
logger.debug("set tpl {}, params: {}".format(k, tvars))
self.seq[int(k)].set_params(tvars)
[docs] def set_template_params(self, d):
"""Sets template params from dictionary"""
self.current_tpl_params.update(d)
assert len(d.keys()) == len(self.seq)
self._set_parameters_from_dict(d)
def _make_sequence(self, *args, **kw):
self.seq = [
Template.create(d, *args, **kw) for d in self.content["templates"]
]
if self.current_tpl_params:
logger.info("apply current tpl params")
self._set_parameters_from_dict(self.current_tpl_params)
return self
# pylint: disable=arguments-differ
[docs] @staticmethod
def create(f, *args, **kw):
"""Creates a :class:`ObservingBlock` node
Args:
f: JSON file with OB definition
Keyword Args:
id: Node id
name: node name
"""
a = ObservingBlock(f, **kw)
logger.debug("Create ObservingBlock -- sn: %d", a.serial_number)
a._make_sequence(*args, **kw) # pylint: disable=protected-access
return a
@property
def module(self):
"""Returns OB filename"""
return "OB_{}".format(self.fname)
@property
def doc(self):
"""Returns OB name from the corresponding kw"""
return "OB:{}".format(self.descr["name"])
[docs] def clone(self, *args, **kw):
"""Fake ctor for OB nodes"""
self.serial_number = Counter.new_value()
return self._make_sequence(*args, **kw)