mirror of
https://github.com/bec-project/bec.git
synced 2026-06-02 00:08:31 +02:00
refactor: rename user scripts to user macros
This commit is contained in:
Klaus Wakonig
@@ -30,7 +30,7 @@ from bec_lib.logger import bec_logger
|
||||
from bec_lib.plugin_helper import _get_available_plugins
|
||||
from bec_lib.scan_history import ScanHistory
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from bec_lib.user_scripts_mixin import UserScriptsMixin
|
||||
from bec_lib.user_macros_mixin import UserMacrosMixin
|
||||
from bec_lib.utils.import_utils import lazy_import_from
|
||||
|
||||
logger = bec_logger.logger
|
||||
@@ -90,7 +90,7 @@ class LiveUpdatesConfig(BaseModel):
|
||||
print_client_messages: bool = True
|
||||
|
||||
|
||||
class BECClient(BECService, UserScriptsMixin):
|
||||
class BECClient(BECService, UserMacrosMixin):
|
||||
"""
|
||||
The BECClient class is the main entry point for the BEC client and all derived classes.
|
||||
"""
|
||||
@@ -212,7 +212,7 @@ class BECClient(BECService, UserScriptsMixin):
|
||||
self._start_device_manager()
|
||||
self._start_scan_queue()
|
||||
self._start_alarm_handler()
|
||||
self.load_all_user_scripts()
|
||||
self.load_all_user_macros()
|
||||
self.config = self.device_manager.config_helper
|
||||
self.history = ScanHistory(client=self)
|
||||
self.dap = DAPPlugins(self)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
This module provides a mixin class for the BEC class that allows the user to load and unload scripts from the `scripts` directory.
|
||||
This module provides a mixin class for the BEC class that allows the user to load and unload macros from the `macros` directory.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -28,111 +28,111 @@ pylint = lazy_import("pylint")
|
||||
CollectingReporter = lazy_import_from("pylint.reporters", ("CollectingReporter",))
|
||||
|
||||
|
||||
class UserScriptsMixin:
|
||||
class UserMacrosMixin:
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._scripts = {}
|
||||
self._macros = {}
|
||||
|
||||
def load_all_user_scripts(self) -> None:
|
||||
def load_all_user_macros(self) -> None:
|
||||
try:
|
||||
self._load_all_user_scripts()
|
||||
self._load_all_user_macros()
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.error(f"Error while loading user scripts: \n {content}")
|
||||
logger.error(f"Error while loading user macros: \n {content}")
|
||||
|
||||
def _load_all_user_scripts(self) -> None:
|
||||
"""Load all scripts from the `scripts` directory.
|
||||
def _load_all_user_macros(self) -> None:
|
||||
"""Load all macros from the `macros` directory.
|
||||
|
||||
Runs a callback of type `EventType.NAMESPACE_UPDATE`
|
||||
to inform clients about added objects in the namesapce.
|
||||
"""
|
||||
self.forget_all_user_scripts()
|
||||
self.forget_all_user_macros()
|
||||
|
||||
# load all scripts from the scripts directory
|
||||
# load all macros from the macros directory
|
||||
current_path = pathlib.Path(__file__).parent.resolve()
|
||||
script_files = glob.glob(os.path.abspath(os.path.join(current_path, "../scripts/*.py")))
|
||||
macro_files = glob.glob(os.path.abspath(os.path.join(current_path, "../macros/*.py")))
|
||||
|
||||
# load all scripts from the user's script directory in the home directory
|
||||
user_script_dir = os.path.join(os.path.expanduser("~"), "bec", "scripts")
|
||||
if os.path.exists(user_script_dir):
|
||||
script_files.extend(glob.glob(os.path.abspath(os.path.join(user_script_dir, "*.py"))))
|
||||
# load all macros from the user's macro directory in the home directory
|
||||
user_macro_dir = os.path.join(os.path.expanduser("~"), "bec", "macros")
|
||||
if os.path.exists(user_macro_dir):
|
||||
macro_files.extend(glob.glob(os.path.abspath(os.path.join(user_macro_dir, "*.py"))))
|
||||
|
||||
# load scripts from the plugins
|
||||
# load macros from the plugins
|
||||
plugins = importlib.metadata.entry_points(group="bec")
|
||||
for plugin in plugins:
|
||||
if plugin.name == "plugin_bec":
|
||||
plugin = plugin.load()
|
||||
plugin_scripts_dir = os.path.join(plugin.__path__[0], "scripts")
|
||||
if os.path.exists(plugin_scripts_dir):
|
||||
script_files.extend(
|
||||
glob.glob(os.path.abspath(os.path.join(plugin_scripts_dir, "*.py")))
|
||||
plugin_macros_dir = os.path.join(plugin.__path__[0], "macros")
|
||||
if os.path.exists(plugin_macros_dir):
|
||||
macro_files.extend(
|
||||
glob.glob(os.path.abspath(os.path.join(plugin_macros_dir, "*.py")))
|
||||
)
|
||||
|
||||
for file in script_files:
|
||||
self.load_user_script(file)
|
||||
builtins.__dict__.update({name: v["cls"] for name, v in self._scripts.items()})
|
||||
for file in macro_files:
|
||||
self.load_user_macro(file)
|
||||
builtins.__dict__.update({name: v["cls"] for name, v in self._macros.items()})
|
||||
|
||||
def forget_all_user_scripts(self) -> None:
|
||||
"""unload / remove loaded user scripts from builtins. Files will remain untouched.
|
||||
def forget_all_user_macros(self) -> None:
|
||||
"""unload / remove loaded user macros from builtins. Files will remain untouched.
|
||||
|
||||
Runs a callback of type `EventType.NAMESPACE_UPDATE`
|
||||
to inform clients about removing objects from the namesapce.
|
||||
|
||||
"""
|
||||
for name, obj in self._scripts.items():
|
||||
for name, obj in self._macros.items():
|
||||
builtins.__dict__.pop(name)
|
||||
self.callbacks.run(
|
||||
EventType.NAMESPACE_UPDATE, action="remove", ns_objects={name: obj["cls"]}
|
||||
)
|
||||
self._scripts.clear()
|
||||
self._macros.clear()
|
||||
|
||||
def load_user_script(self, file: str) -> None:
|
||||
"""load a user script file and import all its definitions
|
||||
def load_user_macro(self, file: str) -> None:
|
||||
"""load a user macro file and import all its definitions
|
||||
|
||||
Args:
|
||||
file (str): Full path to the script file.
|
||||
file (str): Full path to the macro file.
|
||||
"""
|
||||
# TODO: re-enable linter
|
||||
# self._run_linter_on_file(file)
|
||||
module_members = self._load_script_module(file)
|
||||
module_members = self._load_macro_module(file)
|
||||
for name, cls in module_members:
|
||||
if not callable(cls):
|
||||
continue
|
||||
# ignore imported classes
|
||||
if cls.__module__ != "scripts":
|
||||
if cls.__module__ != "macros":
|
||||
continue
|
||||
if name in self._scripts:
|
||||
if name in self._macros:
|
||||
logger.warning(f"Conflicting definitions for {name}.")
|
||||
logger.info(f"Importing {name}")
|
||||
self._scripts[name] = {"cls": cls, "fname": file}
|
||||
self._macros[name] = {"cls": cls, "fname": file}
|
||||
self.callbacks.run(EventType.NAMESPACE_UPDATE, action="add", ns_objects={name: cls})
|
||||
|
||||
def forget_user_script(self, name: str) -> None:
|
||||
"""unload / remove a user scripts. The file will remain on disk."""
|
||||
if name not in self._scripts:
|
||||
logger.error(f"{name} is not a known user script.")
|
||||
def forget_user_macro(self, name: str) -> None:
|
||||
"""unload / remove a user macros. The file will remain on disk."""
|
||||
if name not in self._macros:
|
||||
logger.error(f"{name} is not a known user macro.")
|
||||
return
|
||||
self.callbacks.run(
|
||||
EventType.NAMESPACE_UPDATE,
|
||||
action="remove",
|
||||
ns_objects={name: self._scripts[name]["cls"]},
|
||||
ns_objects={name: self._macros[name]["cls"]},
|
||||
)
|
||||
builtins.__dict__.pop(name)
|
||||
self._scripts.pop(name)
|
||||
self._macros.pop(name)
|
||||
|
||||
def list_user_scripts(self):
|
||||
"""display all currently loaded user functions"""
|
||||
def list_user_macros(self):
|
||||
"""display all currently loaded user macros"""
|
||||
console = Console()
|
||||
table = Table(title="User scripts")
|
||||
table = Table(title="User macros")
|
||||
table.add_column("Name", justify="center")
|
||||
table.add_column("Location", justify="center", overflow="fold")
|
||||
|
||||
for name, content in self._scripts.items():
|
||||
for name, content in self._macros.items():
|
||||
table.add_row(name, content.get("fname"))
|
||||
console.print(table)
|
||||
|
||||
def _load_script_module(self, file) -> list:
|
||||
module_spec = importlib.util.spec_from_file_location("scripts", file)
|
||||
def _load_macro_module(self, file) -> list:
|
||||
module_spec = importlib.util.spec_from_file_location("macros", file)
|
||||
plugin_module = importlib.util.module_from_spec(module_spec)
|
||||
module_spec.loader.exec_module(plugin_module)
|
||||
module_members = inspect.getmembers(plugin_module)
|
||||
@@ -0,0 +1,92 @@
|
||||
import builtins
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from bec_lib.callback_handler import EventType
|
||||
from bec_lib.user_macros_mixin import UserMacrosMixin
|
||||
|
||||
# pylint: disable=no-member
|
||||
# pylint: disable=missing-function-docstring
|
||||
# pylint: disable=redefined-outer-name
|
||||
# pylint: disable=protected-access
|
||||
|
||||
|
||||
def dummy_func():
|
||||
pass
|
||||
|
||||
|
||||
def dummy_func2():
|
||||
pass
|
||||
|
||||
|
||||
class client_user_macros_mixin(UserMacrosMixin):
|
||||
def __init__(self):
|
||||
self.callbacks = None
|
||||
super().__init__()
|
||||
self._macros = {}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def macros():
|
||||
yield client_user_macros_mixin()
|
||||
|
||||
|
||||
def test_user_macros_forget(macros):
|
||||
macros.callbacks = mock.MagicMock()
|
||||
mock_run = macros.callbacks.run
|
||||
macros._macros = {"test": {"cls": dummy_func, "file": "path_to_my_file.py"}}
|
||||
builtins.test = dummy_func
|
||||
macros.forget_all_user_macros()
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="remove", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "test" not in builtins.__dict__
|
||||
assert len(macros._macros) == 0
|
||||
|
||||
|
||||
def test_user_macro_forget(macros):
|
||||
macros.callbacks = mock.MagicMock()
|
||||
mock_run = macros.callbacks.run
|
||||
macros._macros = {"test": {"cls": dummy_func, "file": "path_to_my_file.py"}}
|
||||
builtins.test = dummy_func
|
||||
macros.forget_user_macro("test")
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="remove", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "test" not in builtins.__dict__
|
||||
|
||||
|
||||
def test_load_user_macro(macros):
|
||||
macros.callbacks = mock.MagicMock()
|
||||
mock_run = macros.callbacks.run
|
||||
builtins.__dict__["dev"] = macros
|
||||
dummy_func.__module__ = "macros"
|
||||
with mock.patch.object(macros, "_run_linter_on_file") as linter:
|
||||
with mock.patch.object(
|
||||
macros,
|
||||
"_load_macro_module",
|
||||
return_value=[("test", dummy_func), ("wrong_test", dummy_func2)],
|
||||
) as load_macro:
|
||||
macros.load_user_macro("dummy")
|
||||
assert load_macro.call_count == 1
|
||||
assert load_macro.call_args == mock.call("dummy")
|
||||
assert "test" in macros._macros
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="add", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "wrong_test" not in macros._macros
|
||||
# linter.assert_called_once_with("dummy") #TODO: re-enable this test once issue #298 is fixed
|
||||
|
||||
|
||||
# def test_user_macro_linter():
|
||||
# macros = UsermacrosMixin()
|
||||
# current_path = pathlib.Path(__file__).parent.resolve()
|
||||
# macro_path = os.path.join(current_path, "test_data", "user_macro_with_bug.py")
|
||||
# builtins.__dict__["dev"] = macros
|
||||
# with mock.patch("bec_lib.user_macros_mixin.logger") as logger:
|
||||
# macros._run_linter_on_file(macro_path)
|
||||
# logger.error.assert_called_once()
|
||||
@@ -1,92 +0,0 @@
|
||||
import builtins
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from bec_lib.callback_handler import EventType
|
||||
from bec_lib.user_scripts_mixin import UserScriptsMixin
|
||||
|
||||
# pylint: disable=no-member
|
||||
# pylint: disable=missing-function-docstring
|
||||
# pylint: disable=redefined-outer-name
|
||||
# pylint: disable=protected-access
|
||||
|
||||
|
||||
def dummy_func():
|
||||
pass
|
||||
|
||||
|
||||
def dummy_func2():
|
||||
pass
|
||||
|
||||
|
||||
class client_user_scripts_mixin(UserScriptsMixin):
|
||||
def __init__(self):
|
||||
self.callbacks = None
|
||||
super().__init__()
|
||||
self._scripts = {}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scripts():
|
||||
yield client_user_scripts_mixin()
|
||||
|
||||
|
||||
def test_user_scripts_forget(scripts):
|
||||
scripts.callbacks = mock.MagicMock()
|
||||
mock_run = scripts.callbacks.run
|
||||
scripts._scripts = {"test": {"cls": dummy_func, "file": "path_to_my_file.py"}}
|
||||
builtins.test = dummy_func
|
||||
scripts.forget_all_user_scripts()
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="remove", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "test" not in builtins.__dict__
|
||||
assert len(scripts._scripts) == 0
|
||||
|
||||
|
||||
def test_user_script_forget(scripts):
|
||||
scripts.callbacks = mock.MagicMock()
|
||||
mock_run = scripts.callbacks.run
|
||||
scripts._scripts = {"test": {"cls": dummy_func, "file": "path_to_my_file.py"}}
|
||||
builtins.test = dummy_func
|
||||
scripts.forget_user_script("test")
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="remove", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "test" not in builtins.__dict__
|
||||
|
||||
|
||||
def test_load_user_script(scripts):
|
||||
scripts.callbacks = mock.MagicMock()
|
||||
mock_run = scripts.callbacks.run
|
||||
builtins.__dict__["dev"] = scripts
|
||||
dummy_func.__module__ = "scripts"
|
||||
with mock.patch.object(scripts, "_run_linter_on_file") as linter:
|
||||
with mock.patch.object(
|
||||
scripts,
|
||||
"_load_script_module",
|
||||
return_value=[("test", dummy_func), ("wrong_test", dummy_func2)],
|
||||
) as load_script:
|
||||
scripts.load_user_script("dummy")
|
||||
assert load_script.call_count == 1
|
||||
assert load_script.call_args == mock.call("dummy")
|
||||
assert "test" in scripts._scripts
|
||||
assert mock_run.call_count == 1
|
||||
assert mock_run.call_args == mock.call(
|
||||
EventType.NAMESPACE_UPDATE, action="add", ns_objects={"test": dummy_func}
|
||||
)
|
||||
assert "wrong_test" not in scripts._scripts
|
||||
# linter.assert_called_once_with("dummy") #TODO: re-enable this test once issue #298 is fixed
|
||||
|
||||
|
||||
# def test_user_script_linter():
|
||||
# scripts = UserScriptsMixin()
|
||||
# current_path = pathlib.Path(__file__).parent.resolve()
|
||||
# script_path = os.path.join(current_path, "test_data", "user_script_with_bug.py")
|
||||
# builtins.__dict__["dev"] = scripts
|
||||
# with mock.patch("bec_lib.user_scripts_mixin.logger") as logger:
|
||||
# scripts._run_linter_on_file(script_path)
|
||||
# logger.error.assert_called_once()
|
||||
@@ -298,20 +298,20 @@ Type: function
|
||||
```
|
||||
The shell printout provides information about the scan signature, parameters, as well as a syntax example at the bottom.
|
||||
|
||||
### How to write a script
|
||||
### How to write a macro
|
||||
-----------------------
|
||||
|
||||
Scripts are user defined functions that can be executed from the BEC console (CLI).
|
||||
They are stored in the ``scripts`` folder and can be edited with any text editor.
|
||||
The scripts are loaded automatically on startup of the BEC console but can also be reloaded by typing ``bec.load_all_user_scripts()`` in the command-line.
|
||||
This command will load scripts from three locations:
|
||||
Macros are user defined functions that can be executed from the BEC console (CLI).
|
||||
They are stored in the ``macros`` folder and can be edited with any text editor.
|
||||
The macros are loaded automatically on startup of the BEC console but can also be reloaded by typing ``bec.load_all_user_macros()`` in the command-line.
|
||||
This command will load macros from three locations:
|
||||
|
||||
1. from `~/bec/scripts/` in your home directory,
|
||||
1. from the beamline plugin directory, e.g. `/csaxs_bec/csaxs_bec/scripts/`
|
||||
1. from `bec/bec_lib/scripts/` (only useful if you have the entire source code of BEC installed locally).
|
||||
1. from `~/bec/macros/` in your home directory,
|
||||
1. from the beamline plugin directory, e.g. `/csaxs_bec/csaxs_bec/macros/`
|
||||
1. from `bec/bec_lib/macros/` (only useful if you have the entire source code of BEC installed locally).
|
||||
|
||||
|
||||
An example of a user script could be a function to move a specific motor to a predefined position:
|
||||
An example of a user macro could be a function to move a specific motor to a predefined position:
|
||||
|
||||
```python
|
||||
def samx_in():
|
||||
@@ -340,7 +340,7 @@ A slightly more complex example could be a sequence of scans that are executed i
|
||||
close_shutter()
|
||||
```
|
||||
|
||||
This script can be executed by typing ``overnight_scan()`` in the BEC console and would execute the following sequence of commands:
|
||||
This macro can be executed by typing ``overnight_scan()`` in the BEC console and would execute the following sequence of commands:
|
||||
|
||||
1. Open the shutter
|
||||
2. Move the sample in
|
||||
@@ -348,6 +348,12 @@ This script can be executed by typing ``overnight_scan()`` in the BEC console an
|
||||
4. Move the sample out
|
||||
5. Close the shutter
|
||||
|
||||
Macros can be very useful to automate repetitive tasks or to create complex sequences of commands that can be executed with a single command.
|
||||
|
||||
```{important}
|
||||
As macros are loaded automatically, they should not contain any command outside of a function definition to avoid executing code unintentionally.
|
||||
```
|
||||
|
||||
### Create a custom scan
|
||||
|
||||
As seen above, scans can be access through `scans.`.
|
||||
|
||||
Reference in New Issue
Block a user