"""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
from datetime import datetime
from pathlib import Path
import attr
from .sequence import Sequence
from .template import Template
from ..counter import Counter
from .. import ob
from ..exc import SeqException
from .state import T_STATE
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)
from_otto = attr.ib(default=False, kw_only=True)
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
# check file exists and size > 0
p = Path(self.fname)
if not p.exists():
raise SeqException(
None,
None,
"File does not exists, can't load: %s" % self.fname,
) from None
if p.stat().st_size == 0:
raise SeqException(
None,
None,
"Invalid file content, can't load: %s" % self.fname,
) from None
# load and p
with open(self.fname) as fd:
self.content = json.load(fd)
self.descr = self.content.get("obsDescription", {})
name = self.descr.get("name", "OB name")
self.name = "{}({})".format(name, self.content.get("obId", "eng"))
def _make_sequence(self, *args, parent=None, **kw):
self.in_error = False
self.state = T_STATE.NOT_STARTED
self.seq = [
Template.create(d, parent, *args, **kw)
for d in self.content["templates"]
]
return self
def _publish_state(self):
c = ob.OB.controller.get()
if c:
c.notify_state_change(
self,
datetime.now().isoformat(),
obId=self.content.get("obId", ""),
otto_report=self.from_otto,
)
# 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)
kw.pop("from_otto", False)
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.get("name", "<name>"), self.descr.get("obId", "ID")
)
[docs] def clone(self, *args, **kw):
"""Fake ctor for OB nodes"""
self.serial_number = Counter.new_value()
return self._make_sequence(*args, **kw)