[writer] filename as template and no aliases fix

nexus_writer_service/subscribers/devices.py

@@ -56,10 +56,13 @@ def shortnamemap(names, separator=":"):
     """
     Map full Redis names to short (but still unique) names
 
-    :param lst(str) names:
+    :param list(str) names:
     :param str separator:
     :returns dict:
     """
+    if not names:
+        return {}
+    names = set(names)
     parts = [name.split(separator) for name in names]
     nparts = max(map(len, parts))
     parts = [([""] * (nparts - len(lst))) + lst for lst in parts]

nexus_writer_service/utils/scan_utils.py

@@ -121,11 +121,13 @@ def current_internal_filename(scan_saving=None):
     basename = config_utils.scan_saving_get(
         "data_filename", "", scan_saving=scan_saving
     )
-    if basename:
-        basename = os.path.splitext(basename)[0]
-        return os.path.join(basename + ".h5")
-    else:
-        return ""
+    if not basename:
+        return basename
+    attrs = config_utils.scan_saving_attrs(template=basename, scan_saving=scan_saving)
+    try:
+        return basename.format(**attrs) + ".h5"
+    except KeyError as e:
+        raise RuntimeError("Missing '{}' attribute in SCAN_SAVING".format(e))
 
 
 def filename_int2ext(filename=None, **overwrite):

tests/nexus_writer/conftest.py

@@ -77,7 +77,9 @@ def nexus_config_session(beacon, lima_simulator, lima_simulator2):
 def scan_saving_nopolicy(session, scan_tmpdir):
     scan_saving = session.scan_saving
     scan_saving.base_path = str(scan_tmpdir)
-    scan_saving.data_filename = "dataset"
+    scan_saving.data_filename = "{a}_{b}"
+    scan_saving.add("a", "a")
+    scan_saving.add("b", "b")
     measurementgroup.set_active_name(nxw_test_config.technique["withoutpolicy"] + "MG")
 
 
@@ -85,8 +87,8 @@ def scan_saving_policy(session, scan_tmpdir):
     scan_saving = session.scan_saving
     data_policy.newtmpexperiment("prop123", root=str(scan_tmpdir))
     scan_saving.add("sample", "sample")
-    scan_saving.add("technique", nxw_test_config.technique["withpolicy"])
     scan_saving.add("dataset", "dataset")
+    scan_saving.add("technique", nxw_test_config.technique["withpolicy"])
     measurementgroup.set_active_name(scan_saving.technique + "MG")
 
 

tests/nexus_writer/test_devices.py

+# -*- coding: utf-8 -*-
+#
+# This file is part of the bliss project
+#
+# Copyright (c) 2015-2019 Beamline Control Unit, ESRF
+# Distributed under the GNU LGPLv3. See LICENSE for more info.
+
+from nexus_writer_service.subscribers import devices
+
+
+def test_devices_shortnamemap():
+    assert devices.shortnamemap([]) == {}
+    assert devices.shortnamemap(["a:b:c"]) == {"a:b:c": "c"}
+    assert devices.shortnamemap(["a:b:c", "a:b:d"]) == {"a:b:c": "c", "a:b:d": "d"}
+    assert devices.shortnamemap(["a:b:c", "a:b:c", "a:b:d"]) == {
+        "a:b:c": "c",
+        "a:b:d": "d",
+    }
+    assert devices.shortnamemap(["a:b:c", "a:b:d", "c"]) == {
+        "a:b:c": "b:c",
+        "a:b:d": "d",
+        "c": "c",
+    }
+    assert devices.shortnamemap(["a:b:c", "b:c:d", "b:c"]) == {
+        "a:b:c": "a:b:c",
+        "b:c:d": "d",
+        "b:c": "b:c",
+    }
+    assert devices.shortnamemap(["a:b:c", "a:c"]) == {"a:b:c": "b:c", "a:c": "a:c"}
+    assert devices.shortnamemap(["a:b:c", "b:c"]) == {"a:b:c": "a:b:c", "b:c": "b:c"}
+    assert devices.shortnamemap(["b:a", "c:a"]) == {"b:a": "b:a", "c:a": "c:a"}

tests/nexus_writer/test_writer_config.py

@@ -23,9 +23,7 @@ def test_config_withoutpolicy(nexus_base_session_nopolicy):
     expected_directory = os.path.join(scan_tmpdir, session.name)
     assert directory == expected_directory
     filenames = scan_utils.current_filenames()
-    expected_filenames = [
-        os.path.join(scan_tmpdir, session.name, "dataset_external.h5")
-    ]
+    expected_filenames = [os.path.join(scan_tmpdir, session.name, "a_b_external.h5")]
     assert filenames == expected_filenames