Commit 06985c6e authored by Sebastien Petitdemange's avatar Sebastien Petitdemange

Merge branch 'issue-454' into 'master'

Issue 454

See merge request !957
parents 1584cb3e c4029bfa
Pipeline #6779 failed with stages
in 19 minutes and 2 seconds
......@@ -31,7 +31,7 @@ from bliss.common.motor_group import Group
from bliss.common.cleanup import cleanup, axis as cleanup_axis
from bliss.common.axis import estimate_duration
from bliss.scanning.default import DefaultAcquisitionChain
from bliss.scanning import scan as scan_module
from bliss.scanning.scan import Scan, StepScanDataWatch
from bliss.scanning.acquisition.motor import VariableStepTriggerMaster
from bliss.scanning.acquisition.motor import (
LinearStepTriggerMaster,
......@@ -43,22 +43,6 @@ _log = logging.getLogger("bliss.scans")
DEFAULT_CHAIN = DefaultAcquisitionChain()
def step_scan(chain, scan_info, name=None, save=True, save_images=True):
scan_data_watch = scan_module.StepScanDataWatch()
config = scan_module.ScanSaving().get()
writer = config.get("writer") if save else None
if writer:
writer._save_images = save_images
return scan_module.Scan(
chain,
name=name,
parent=config["parent"],
scan_info=scan_info,
writer=writer,
data_watch_callback=scan_data_watch,
)
def ascan(motor, start, stop, npoints, count_time, *counter_args, **kwargs):
"""
Absolute scan
......@@ -138,12 +122,13 @@ def ascan(motor, start, stop, npoints, count_time, *counter_args, **kwargs):
"Scanning %s from %f to %f in %d points", motor.name, start, stop, npoints
)
scan = step_scan(
scan = Scan(
chain,
scan_info,
scan_info=scan_info,
name=kwargs.setdefault("name", "ascan"),
save=scan_info["save"],
save_images=save_images,
data_watch_callback=StepScanDataWatch(),
)
if kwargs.get("run", True):
......@@ -335,12 +320,13 @@ def amesh(
npoints2,
)
scan = step_scan(
scan = Scan(
chain,
scan_info,
scan_info=scan_info,
name=kwargs.setdefault("name", "amesh"),
save=scan_info["save"],
save_images=save_images,
data_watch_callback=StepScanDataWatch(),
)
if kwargs.get("run", True):
......@@ -524,12 +510,13 @@ def a2scan(
npoints,
)
scan = step_scan(
scan = Scan(
chain,
scan_info,
scan_info=scan_info,
name=kwargs.setdefault("name", "a2scan"),
save=scan_info["save"],
save_images=save_images,
data_watch_callback=StepScanDataWatch(),
)
if kwargs.get("run", True):
......@@ -695,12 +682,13 @@ def timescan(count_time, *counter_args, **kwargs):
chain = DEFAULT_CHAIN.get(scan_info, counter_args)
scan = step_scan(
scan = Scan(
chain,
scan_info,
scan_info=scan_info,
name=kwargs.setdefault("name", "timescan"),
save=scan_info["save"],
save_images=save_images,
data_watch_callback=StepScanDataWatch(),
)
if kwargs.get("run", True):
......@@ -846,12 +834,13 @@ def pointscan(motor, positions, count_time, *counter_args, **kwargs):
npoints,
)
scan = step_scan(
scan = Scan(
chain,
scan_info,
scan_info=scan_info,
name=kwargs.setdefault("name", "pointscan"),
save=scan_info["save"],
save_images=save_images,
data_watch_callback=StepScanDataWatch(),
)
scan.run()
......
......@@ -360,7 +360,7 @@ class Session(object):
from bliss.scanning.scan import ScanSaving, ScanDisplay, SCANS
env_dict["SCANS"] = SCANS
env_dict["SCAN_SAVING"] = ScanSaving()
env_dict["SCAN_SAVING"] = ScanSaving(self.name)
env_dict["SCAN_DISPLAY"] = ScanDisplay()
from bliss.common.measurementgroup import ACTIVE_MG
......
......@@ -8,7 +8,6 @@
from bliss.common.utils import add_property
from bliss.common.tango import DeviceProxy, DevFailed
from bliss.common.measurement import SamplingCounter
from bliss.scanning.scan import ScanSaving
from bliss.config.settings import SimpleSetting
from bliss.common import Actuator
import gevent
......@@ -32,12 +31,8 @@ class BpmGroupedReadHandler(SamplingCounter.GroupedReadHandler):
elif self.controller.is_live():
self.__back_to_live = True
self.controller.stop()
# save image if image counter is present
if any([isinstance(c, BpmImage) for c in counters]):
self.controller.save_images(True)
def stop(self, *counters):
self.controller.save_images(False)
if self.__back_to_live:
while self.controller.is_acquiring():
gevent.idle()
......@@ -45,10 +40,7 @@ class BpmGroupedReadHandler(SamplingCounter.GroupedReadHandler):
def read(self, *counters):
result = self.controller.tango_proxy.GetPosition()
return [
cnt.count if isinstance(cnt, BpmImage) else result[cnt.index]
for cnt in counters
]
return [result[cnt.index] for cnt in counters]
class BpmCounter(SamplingCounter):
......@@ -236,19 +228,6 @@ class tango_bpm(object):
def is_out(self):
return self.__control.YagStatus == "out"
def save_images(self, save):
if save:
scan_saving = ScanSaving()
directory = scan_saving.get_path()
image_acq_counter_setting = SimpleSetting(
self.name + ".image", None, int, int, default_value=0
)
image_acq_counter_setting += 1
prefix = self.name + "_image_%d_" % image_acq_counter_setting.get()
self.__control.EnableAutoSaving([directory, prefix])
else:
self.__control.DisableAutoSaving()
def __repr__(self):
try:
msg = (
......
......@@ -163,7 +163,7 @@ class DataNodeIterator(object):
if filter is None or self.node.type in filter:
yield self.node
data_node_2_children = self._get_children_of_children(db_name)
data_node_2_children = self._get_grandchildren(db_name)
all_nodes_names = list()
for children_name in data_node_2_children.values():
all_nodes_names.extend(children_name)
......@@ -207,7 +207,7 @@ class DataNodeIterator(object):
):
yield n
def _get_children_of_children(self, db_name):
def _get_grandchildren(self, db_name):
# grouped all redis request here and cache them
# get all children queue
children_queue = [
......@@ -245,7 +245,7 @@ class DataNodeIterator(object):
pass
else:
db_name = self.node.db_name
data_node_2_children = self._get_children_of_children(db_name)
data_node_2_children = self._get_grandchildren(db_name)
self.last_child_id = {
db_name: len(children)
for db_name, children in data_node_2_children.items()
......
......@@ -746,6 +746,9 @@ class AcquisitionChain(object):
self._device2one_shot_flag = weakref.WeakKeyDictionary()
self._statistic_container = dict()
def trace(self, on):
_logger.setLevel("DEBUG" if on else 0)
@property
def nodes_list(self):
nodes_gen = self._tree.expand_tree()
......
This diff is collapsed.
# -*- coding: utf-8 -*-
#
# This file is part of the bliss project
#
# Copyright (c) 2016 Beamline Control Unit, ESRF
# Distributed under the GNU LGPLv3. See LICENSE for more info.
import numpy
def peak(x, y):
return x[y.argmax()]
def com(x, y):
return numpy.sum(x * y) / numpy.sum(y)
def cen(x, y):
half_val = (max(y) + min(y)) / 2.
nb_value = len(x)
index_above_half = numpy.where(y >= half_val)[0]
slope = numpy.gradient(y, x)
if index_above_half[0] != 0 and index_above_half[-1] != (nb_value - 1):
# standard peak
if len(index_above_half) == 1: # only one point above half_value
indexes = [index_above_half[0] - 1, index_above_half[0] + 1]
else:
indexes = [index_above_half[0], index_above_half[-1]]
elif index_above_half[0] == 0 and index_above_half[-1] == (nb_value - 1):
index_below_half = numpy.where(y <= half_val)[0]
if len(index_below_half) == 1:
indexes = [index_below_half[0] - 1, index_below_half[0] + 1]
else:
indexes = [index_below_half[0], index_below_half[-1]]
elif index_above_half[0] == 0: # falling edge
indexes = [index_above_half[-1]]
else: # raising edge
indexes = [index_above_half[0]]
fwhms = numpy.array([x[i] + ((half_val - y[i]) / slope[i]) for i in indexes])
fwhm = fwhms.max() - fwhms.min()
cfwhm = fwhms.mean()
return cfwhm, fwhm
......@@ -67,15 +67,17 @@ class CsvMasterFile(object):
class Writer(FileWriter):
def __init__(self, root_path, images_root_path, **keys):
def __init__(self, root_path, images_path, data_filename, **keys):
FileWriter.__init__(
self,
root_path,
images_root_path,
master_event_callback=None,
device_event_callback=on_event,
images_path,
data_filename,
master_event_receiver=None,
device_event_receiver=on_event,
**keys
)
def new_master(self, master, scan_file_dir):
return CsvMasterFile(os.path.join(scan_file_dir, master.name + "_master.csv"))
def new_master(self, master):
filename = self.data_filename + "_" + master.name + ".csv"
return CsvMasterFile(os.path.join(self.root_path, filename))
......@@ -44,6 +44,7 @@ class FileWriter(object):
self,
root_path,
images_root_path,
data_filename,
master_event_callback=None,
device_event_callback=None,
**keys
......@@ -51,21 +52,31 @@ class FileWriter(object):
""" A default way to organize file structure
"""
self._save_images = True
self._root_path = root_path
self._images_root_path = images_root_path
self._root_path_template = root_path
self._data_filename_template = data_filename
self._template_dict = {}
self._images_root_path_template = images_root_path
self._master_event_callback = master_event_callback
self._device_event_callback = device_event_callback
self._event_receivers = list()
self.log = logging.getLogger(type(self).__name__)
@property
def template(self):
return self._template_dict
@property
def root_path(self):
return self._root_path
return self._root_path_template.format(**self._template_dict)
@property
def images_root_path(self):
return self._images_root_path
def data_filename(self):
return self._data_filename_template.format(**self._template_dict)
@property
def filename(self):
raise NotImplementedError
def create_path(self, full_path):
try:
......@@ -77,13 +88,13 @@ class FileWriter(object):
raise
def new_scan(self, scan):
self.create_path(self._root_path)
self.new_file(self._root_path, scan.node.name, scan.scan_info)
self.create_path(self.root_path)
self.new_file(scan.node.name, scan.scan_info)
def new_file(self, scan_file_dir, scan_name, scan_info):
def new_file(self, scan_name, scan_info):
pass
def new_master(self, master, scan_file_dir):
def new_master(self, master):
raise NotImplementedError
def add_reference(self, master_entry, referenced_master_entry):
......@@ -122,13 +133,15 @@ class FileWriter(object):
try:
master_entry = master_entries[dev]
except KeyError:
master_entry = self.new_master(dev, scan.path)
master_entry = self.new_master(dev)
master_entries[dev] = master_entry
self._prepare_callbacks(dev, master_entry, self._master_event_callback)
images_path = self._images_root_path.format(
scan=scan.node.name, device=dev.name
images_path = self._images_root_path_template.format(
scan_name=scan.name,
img_acq_device=dev.name,
scan_number=scan.scan_number,
)
self.prepare_saving(dev, images_path)
......@@ -143,7 +156,7 @@ class FileWriter(object):
try:
referenced_master_entry = master_entries[slave]
except KeyError:
referenced_master_entry = self.new_master(slave, scan.path)
referenced_master_entry = self.new_master(slave)
master_entries[slave] = referenced_master_entry
self.add_reference(master_entry, referenced_master_entry)
self._closed = False
......
......@@ -15,11 +15,12 @@ from bliss.scanning.writer.file import FileWriter
class Writer(FileWriter):
def __init__(self, root_path, images_root_path, **keys):
def __init__(self, root_path, images_root_path, data_filename, **keys):
FileWriter.__init__(
self,
root_path,
images_root_path,
data_filename,
master_event_callback=self._on_event,
device_event_callback=self._on_event,
**keys
......@@ -30,9 +31,13 @@ class Writer(FileWriter):
self.measurement = None
self.last_point_index = {}
def new_file(self, scan_file_dir, scan_name, scan_info):
@property
def filename(self):
return os.path.join(self.root_path, self.data_filename + ".h5")
def new_file(self, scan_name, scan_info):
self.close()
self.file = h5py.File(os.path.join(scan_file_dir, "data.h5"))
self.file = h5py.File(self.filename)
self.scan_entry = self.file.create_group(scan_name)
self.scan_entry.attrs["NX_class"] = "NXentry"
scan_title = scan_info.get("title", "untitled")
......@@ -59,7 +64,7 @@ class Writer(FileWriter):
if isinstance(ppos, float):
positioners_dial.create_dataset(pname, dtype="float64", data=ppos)
def new_master(self, master, scan_file_dir):
def new_master(self, master):
return self.measurement.create_group(master.name)
def add_reference(self, master_entry, referenced_master_entry):
......@@ -92,7 +97,7 @@ class Writer(FileWriter):
dataset = parent[channel.fullname]
if not dataset.id.valid:
print("writer is closed. Spurious data point ignored")
print("Writer is closed. Spurious data point ignored")
return
last_point_index = self.last_point_index[channel]
......@@ -116,9 +121,8 @@ class Writer(FileWriter):
self.measurement = None
def get_scan_entries(self):
file_name = os.path.join(self.root_path, "data.h5")
try:
with h5py.File(file_name, mode="r") as f:
with h5py.File(self.filename, mode="r") as f:
return f.keys()
except IOError: # file doesn't exist
return []
......@@ -10,10 +10,10 @@ from bliss.scanning.writer.file import FileWriter
class Writer(FileWriter):
def __init__(self, *args, **keys):
FileWriter.__init__(self, "", "")
FileWriter.__init__(self, "", "", "")
def new_scan(self, scan):
return
pass
def create_path(self, scan_recorder):
return
......@@ -23,3 +23,7 @@ class Writer(FileWriter):
def get_scan_entries(self):
return []
@property
def filename(self):
return "<no saving>"
......@@ -130,7 +130,7 @@ class ScanListener:
HEADER = (
"Total {npoints} points{estimation_str}\n"
+ "{not_shown_counters_str}\n"
+ "Scan {scan_nb} {start_time_str} {root_path} "
+ "Scan {scan_nb} {start_time_str} {filename} "
+ "{session_name} user = {user_name}\n"
+ "{title}\n\n"
+ "{column_header}"
......@@ -142,6 +142,7 @@ class ScanListener:
dispatcher.connect(self.__on_scan_new, "scan_new", scan)
dispatcher.connect(self.__on_scan_data, "scan_data", scan)
dispatcher.connect(self.__on_scan_end, "scan_end", scan)
self.real_motors = []
def __on_scan_new(self, scan_info):
scan_type = scan_info.get("type")
......@@ -154,9 +155,6 @@ class ScanListener:
if nb_points is None:
return
if not scan_info["save"]:
scan_info["root_path"] = "<no saving>"
self.col_labels = ["#"]
self.real_motors = []
self.counters = []
......
......@@ -156,7 +156,7 @@ def cli(
)
prompt_label = session_name.upper()
else:
session_id = "unnamed"
session_id = "default"
session_title = u"Bliss shell"
history_filename = ".%s_history" % os.path.basename(sys.argv[0])
prompt_label = "BLISS"
......
......@@ -91,7 +91,7 @@ def init_scans_callbacks(interpreter, output_queue):
interpreter.get_last_client_uuid(),
{
"scan_id": scan_info["node_name"],
"filename": scan_info["root_path"],
"filename": scan_info["filename"],
"scan_actuators": [actuator.name for actuator in scan_actuators],
"npoints": scan_info["npoints"],
"counters": [ct.name for ct in scan_info["counters"]],
......
# Defaults BLISS scans
## BLISS scan functions
## BLISS step-by-steps scan functions
BLISS provides functions to perform scans a user would need for usual
step-by-step measurements.
The goal of all these functions is:
* to create an *acquisition chain*
* to forge a *title* and *type* (if not provided) based on scan parameters
* to check and to define *scan parameters* (motors, number of points, etc.)
These objects are then passed to `step_scan()` function to create a
`bliss.scanning.scan.Scan` object which can be run or returned to
caller. This class publishes data and triggers the file writer
(hdf5, cvs, SpecFile...) if any.
BLISS provides functions to perform step-by-step scans. The acquisition
chain for those scans is built using the `DefaultChain` class.
{% dot scan_dep.svg
......
......@@ -46,7 +46,7 @@ a dictionary, whose key `root_path` is the final path to scan files.
* `base_path`: the highest level directory for the file path, e.g. `/data`
* `user_name`: the current Unix user name
* `session`: current BLISS session name, or `unnamed` if session has no name
* `session`: current BLISS session name
* `template`: defaults to `{session}/`
* `.add(key, value)`: add a new key (string) to the SCAN_SAVING structure
- value can be a scalar or a function
......
......@@ -167,8 +167,9 @@ def test_images_dir_prefix_saving(beacon, lima_simulator, scan_tmpdir, session):
scan_saving.base_path = str(scan_tmpdir)
scan_saving.template = "test"
scan_saving.images_path_template = "{scan}/toto"
scan_saving.images_prefix = "{device}"
scan_saving.images_path_template = "{scan_name}_{scan_number}/toto"
scan_saving.images_prefix = "{img_acq_device}"
scan_saving.scan_number_format = "%1d"
try:
scan_config = scan_saving.get()
......@@ -204,8 +205,9 @@ def test_images_dir_prefix_saving_absolute(
scan_saving.base_path = str(scan_tmpdir)
scan_saving.template = "test"
scan_saving.images_path_relative = False
scan_saving.images_path_template = "{base_path}/test/{scan}/toto"
scan_saving.images_prefix = "{device}"
scan_saving.images_path_template = "{base_path}/test/{scan_name}_{scan_number}/toto"
scan_saving.images_prefix = "{img_acq_device}"
scan_saving.scan_number_format = "%1d"
try:
scan_config = scan_saving.get()
......@@ -213,7 +215,9 @@ def test_images_dir_prefix_saving_absolute(
scan_saving.base_path, scan_saving.template
)
assert scan_config["images_path"] == os.path.join(
scan_saving.base_path, scan_saving.template, "{scan}/toto/{device}"
scan_saving.base_path,
scan_saving.template,
"{scan_name}_{scan_number}/toto/{img_acq_device}",
)
setup_globals.timescan(0.1, simulator, npoints=1)
......
......@@ -62,7 +62,7 @@ def test_software_position_trigger_master(beacon):
DebugMotorMockupAcquisitionDevice("debug", robz),
)
# Run scan
s = Scan(chain, writer=None)
s = Scan(chain, save=False)
with gevent.Timeout(5):
s.run()
# Check data
......@@ -97,7 +97,7 @@ def test_multi_top_master(beacon, diode_acq_device_factory, diode):
scan_params = {"npoints": 0, "count_time": count_time * 2.}
chain.append(DEFAULT_CHAIN.get(scan_params, (diode2,)))
scan = Scan(chain, name="multi_master", writer=None)
scan = Scan(chain, name="multi_master", save=False)
scan.run()
# should be about the same sampling rate
# just to test that both top master run in parallel
......@@ -117,7 +117,7 @@ def test_interrupted_scan(beacon, diode_acq_device_factory):
chain.add(master, acquisition_device_1)
chain.add(master, acquisition_device_2)
# Run scan
s = Scan(chain, writer=None)
s = Scan(chain, save=False)
scan_task = gevent.spawn(s.run)
gevent.sleep(0.2)
......@@ -140,7 +140,7 @@ def test_scan_too_fast(beacon, diode_acq_device_factory):
acquisition_device_1 = diode_acq_device_factory.get(count_time=0.1, npoints=5)
master = SoftwarePositionTriggerMaster(robz, 0, 1, 5)
chain.add(master, acquisition_device_1)
s = Scan(chain, writer=None)
s = Scan(chain, save=False)
with gevent.Timeout(6):
with pytest.raises(RuntimeError) as e_info:
# aborted due to bad triggering on slaves
......@@ -163,7 +163,7 @@ def test_scan_failure(beacon, diode_acq_device_factory):
chain.add(master, acquisition_device_2)
# Run scan
s = Scan(chain, writer=None)
s = Scan(chain, save=False)
with pytest.raises(RuntimeError) as e_info:
s.run()
......
......@@ -21,7 +21,7 @@ from bliss.scanning.chain import AcquisitionChain
@pytest.fixture
def scan_saving():
ss = ScanSaving()
ss = ScanSaving("test")
prev_template = ss.template
yield ss
ss.template = prev_template
......@@ -33,7 +33,7 @@ def test_scan_saving(beacon, scan_saving):
assert parent_node.name == "toto"
assert parent_node.parent is not None
assert parent_node.parent.name == scan_saving.session
assert parent_node.parent.db_name == scan_saving.session + ":" + scan_saving.session
assert parent_node.parent.db_name == scan_saving.session
assert parent_node.db_name == "%s:%s" % (parent_node.parent.db_name, "toto")
scan_saving.template = "toto"
......@@ -49,13 +49,16 @@ def test_scan_saving(beacon, scan_saving):
== """\
Parameters (default)
.base_path = '/tmp/scans'
.data_filename = 'data'
.date = '{date}'
.date_format = '%Y%m%d'
.device = '<images_* only> acquisition device name'
.images_path_relative = True
.images_path_template = '{{scan}}'
.images_prefix = '{{device}}_'
.scan = '<images_* only> scan node name'
.images_path_template = 'scan{{scan_number}}'
.images_prefix = '{{img_acq_device}}_'
.img_acq_device = '<images_* only> acquisition device name'
.scan_name = 'scan name'
.scan_number = 'scan number'
.scan_number_format = '%04d'
.session = '{session}'
.template = 'toto'
.user_name = '{user_name}'
......@@ -109,7 +112,7 @@ def test_simple_continuous_scan_with_session_watcher(session, scan_saving):
)
try:
gevent.sleep(0.1) # wait a bit to have session watcher greenlet started
scan = Scan(chain, parent=scan_saving.get_parent_node(), writer=None)
scan = Scan(chain, save=False)
scan.run()
finally:
session_watcher.kill()
......
......@@ -72,9 +72,8 @@ def test_hdf5_metadata(beacon, session):
s.scan_info["start_timestamp"]
).isoformat()
scan_file = os.path.join(s.path, "data.h5")
with h5py.File(scan_file, "r") as f:
dataset = f[s.name]
with h5py.File(s.writer.filename, "r") as f:
dataset = f[s.node.name]
assert dataset["title"].value == u"ascan roby 0 10 10 0.01"
assert dataset["start_time"].value.startswith(iso_start_time)
assert dataset["measurement"]
......@@ -102,9 +101,7 @@ def test_hdf5_file_items(beacon, session):
return_scan=True,
)
scan_file = os.path.join(s.path, "data.h5")
scan_dump = h5dump(scan_file)
scan_dump = h5dump(s.writer.filename)
ref_ascan_dump = ascan_dump.split("\n")
......@@ -117,10 +114,10 @@ def test_hdf5_file_items(beacon, session):
if in_positioner:
continue
else:
in_scan = l == s.name or l.startswith(s.name + "/")
in_scan = l == s.node.name or l.startswith(s.node.name + "/")
if in_scan:
if l.startswith(s.name + "/measurement/group_"):
group_name = l.replace(s.name + "/measurement/", "").split("/")[0]
if l.startswith(s.node.name + "/measurement/group_"):
group_name = l.replace(s.node.name + "/measurement/", "").split("/")[0]
else:
continue
if "positioner" in l:
......@@ -128,13 +125,13 @@ def test_hdf5_file_items(beacon, session):
continue
else:
in_positioner = False
assert l == ref_ascan_dump[i].format(ascan=s.name, group_name=group_name)
assert l == ref_ascan_dump[i].format(ascan=s.node.name, group_name=group_name)
i += 1
f = h5py.File(scan_file)
f = h5py.File(s.writer.filename)
assert (
f[f[s.name]["measurement"][group_name]["timer"].value]