controller_settings: first version

uniform configuration parameters with settings parameters.

bliss/common/controller_settings.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.
+
+import yaml
+from bliss.config.settings import HashObjSetting
+from bliss.config.conductor.client import remote_open
+
+from .utils import autocomplete_property
+
+
+def _find_dict(name, d):
+    if d.get("name") == name:
+        return d
+    for key, value in d.items():
+        if isinstance(value, dict):
+            sub_dict = _find_dict(name, value)
+        elif isinstance(value, list):
+            sub_dict = _find_list(name, value)
+        else:
+            continue
+
+        if sub_dict is not None:
+            return sub_dict
+
+
+def _find_list(name, l):
+    for value in l:
+        if isinstance(value, dict):
+            sub_dict = _find_dict(name, value)
+        elif isinstance(value, list):
+            sub_dict = _find_list(name, value)
+        else:
+            continue
+        if sub_dict is not None:
+            return sub_dict
+
+
+class ControllerConfigSettings:
+    def __init__(self, config, settings_keys=list(), default_config_values=dict()):
+        try:
+            self.__name = config["name"]
+        except KeyError:
+            raise RuntimeError("config object must have a name")
+
+        self.__default_config_values = default_config_values
+        self.__settings_keys = settings_keys
+        self._config = config
+        self.__update_settings()
+
+    @autocomplete_property
+    def config(self):
+        return self._config
+
+    @autocomplete_property
+    def settings(self):
+        return self._settings
+
+    def __update_settings(self):
+        config = self.config
+        defaults = self.__default_config_values.copy()
+        default_values = {
+            key: config.get(key, defaults.get(key)) for key in self.__settings_keys
+        }
+        self._settings = HashObjSetting(
+            f"{self.__name}:settings", default_values=default_values
+        )
+
+    def apply_config(self, reload=False):
+        if reload:
+            with remote_open(self.config.filename) as f:
+                d = yaml.safe_load(f.read())
+            if isinstance(d, dict):
+                d = _find_dict(self.__name, d)
+            elif isinstance(d, list):
+                d = _find_list(self.__name, d)
+            else:
+                d = None
+
+            if d is None:
+                raise RuntimeError(
+                    f"Can't find config node named:{self.__name} "
+                    f"in file:{self.config.filename}"
+                )
+            self.config.update(d)
+        self._settings.remove(*self.__settings_keys)
+        self.__update_settings()

tests/controllers_sw/test_controller_settings.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.
+import pytest
+from bliss.common.controller_settings import ControllerConfigSettings
+
+
+@pytest.mark.parametrize(
+    "ctrl_name",
+    [
+        "controller_mixin1",
+        "controller_mixin2",
+        "controller_mixin3",
+        "controller_mixin4",
+    ],
+)
+def test_controller_settings_basic_check(beacon, ctrl_name):
+    cfg = beacon.get(ctrl_name)
+    settings_keys = [name for name in cfg if name != "name"]
+    ctrl = ControllerConfigSettings(cfg, settings_keys=settings_keys)
+    config_dict = {key: ctrl.config.get(key) for key in settings_keys}
+    assert config_dict == dict(ctrl.settings)
+    ctrl.settings["only_in_settings"] = 42
+    config_and_settings = {key: ctrl.settings.get(key) for key in settings_keys}
+    assert config_dict == config_and_settings
+
+
+@pytest.mark.parametrize(
+    "ctrl_name",
+    [
+        ("controller_mixin1"),
+        ("controller_mixin2"),
+        ("controller_mixin3"),
+        ("controller_mixin4"),
+    ],
+)
+def test_controller_settings_apply_config(beacon, ctrl_name):
+    cfg = beacon.get(ctrl_name)
+    settings_keys = [name for name in cfg if name != "name"]
+    ctrl = ControllerConfigSettings(cfg, settings_keys=settings_keys)
+    d = {key: value * 2 for key, value in ctrl.settings.items()}
+    ctrl.settings.update(d)
+    config_dict = {key: ctrl.config.get(key) for key in settings_keys}
+    assert config_dict != dict(ctrl.settings)
+    ctrl.apply_config()
+    assert config_dict == dict(ctrl.settings)
+
+
+@pytest.mark.parametrize(
+    "ctrl_name",
+    [
+        ("controller_mixin1"),
+        ("controller_mixin2"),
+        ("controller_mixin3"),
+        ("controller_mixin4"),
+    ],
+)
+def test_controller_settings_apply_config_reload(beacon, ctrl_name):
+    cfg = beacon.get(ctrl_name)
+    settings_keys = [name for name in cfg if name != "name"]
+    ctrl = ControllerConfigSettings(cfg, settings_keys=settings_keys)
+
+    previous_config_dict = {key: ctrl.config.get(key) for key in settings_keys}
+    current_config_dict = {key: ctrl.config.get(key) * 2 for key in settings_keys}
+    cfg.update(current_config_dict)
+
+    assert previous_config_dict == dict(ctrl.settings)
+    ctrl.apply_config()
+    assert current_config_dict == dict(ctrl.settings)
+
+    ctrl.apply_config(reload=True)
+    assert previous_config_dict == dict(ctrl.settings)

tests/test_configuration/simple_controller.yml

+- ctrl: hello
+  obj:
+    - name: controller_mixin1
+      speed: 10
+      velocity: 1.2
+    - name: controller_mixin2
+      mode: fixed
+      reading_speed: slow
+
+      
\ No newline at end of file

tests/test_configuration/simple_controller2.yml

+ctrl: hello
+first:
+  name: controller_mixin3
+  value: 42
+second:
+  name: controller_mixin4
+  super: mario
\ No newline at end of file