bec/xtreme_bec
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: upgraded to new plugin structure

Browse Source
This commit is contained in:
wakonig_k
2024-05-08 15:17:18 +02:00
parent ad8f227691
commit b82ea591b3
39 changed files with 335 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.git_hooks/post-commit Normal file
View File

@@ -0,0 +1,3 @@
SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
semantic-release changelog -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__
semantic-release version -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__

3
.git_hooks/pre-commit Normal file
View File

@@ -0,0 +1,3 @@
black --line-length=100 $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
isort --line-length=100 --profile=black --multi-line=3 --trailing-comma $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
git add $(git diff --cached --name-only --diff-filter=ACM -- '*.py')

3
.gitignore vendored
View File

@@ -8,6 +8,9 @@
**/.pytest_cache
**/*.egg*
# recovery_config files
recovery_config_*
# file writer data
**.h5

6
.gitlab-ci.yml Normal file
View File

@@ -0,0 +1,6 @@
include:
- file: /templates/plugin-repo-template.yml
inputs:
name: xtreme_bec
target: xtreme_bec
project: bec/awi_utils

1
bec_plugins/__init__.py
View File

@@ -1 +0,0 @@
from .bec_client import *

1
bec_plugins/bec_client/__init__.py
View File

@@ -1 +0,0 @@
from .plugins import *

245
bec_plugins/bec_client/hli/spec_hli.py
View File

@@ -1,245 +0,0 @@
from bec_lib.devicemanager import Device
from bec_lib.scan_report import ScanReport
# pylint:disable=undefined-variable
# pylint: disable=too-many-arguments
def dscan(
motor1: Device, m1_from: float, m1_to: float, steps: int, exp_time: float, **kwargs
) -> ScanReport:
"""Relative line scan with one device.
Args:
motor1 (Device): Device that should be scanned.
m1_from (float): Start position relative to the current position.
m1_to (float): End position relative to the current position.
steps (int): Number of steps.
exp_time (float): Exposure time.
Returns:
ScanReport: Status object.
Examples:
>>> dscan(dev.motor1, -5, 5, 10, 0.1)
"""
return scans.line_scan(
motor1, m1_from, m1_to, steps=steps, exp_time=exp_time, relative=True, **kwargs
)
def d2scan(
motor1: Device,
m1_from: float,
m1_to: float,
motor2: Device,
m2_from: float,
m2_to: float,
steps: int,
exp_time: float,
**kwargs
) -> ScanReport:
"""Relative line scan with two devices.
Args:
motor1 (Device): First device that should be scanned.
m1_from (float): Start position of the first device relative to its current position.
m1_to (float): End position of the first device relative to its current position.
motor2 (Device): Second device that should be scanned.
m2_from (float): Start position of the second device relative to its current position.
m2_to (float): End position of the second device relative to its current position.
steps (int): Number of steps.
exp_time (float): Exposure time
Returns:
ScanReport: Status object.
Examples:
>>> d2scan(dev.motor1, -5, 5, dev.motor2, -8, 8, 10, 0.1)
"""
return scans.line_scan(
motor1,
m1_from,
m1_to,
motor2,
m2_from,
m2_to,
steps=steps,
exp_time=exp_time,
relative=True,
**kwargs
)
def ascan(motor1, m1_from, m1_to, steps, exp_time, **kwargs):
"""Absolute line scan with one device.
Args:
motor1 (Device): Device that should be scanned.
m1_from (float): Start position.
m1_to (float): End position.
steps (int): Number of steps.
exp_time (float): Exposure time.
Returns:
ScanReport: Status object.
Examples:
>>> ascan(dev.motor1, -5, 5, 10, 0.1)
"""
return scans.line_scan(
motor1, m1_from, m1_to, steps=steps, exp_time=exp_time, relative=False, **kwargs
)
def a2scan(motor1, m1_from, m1_to, motor2, m2_from, m2_to, steps, exp_time, **kwargs):
"""Absolute line scan with two devices.
Args:
motor1 (Device): First device that should be scanned.
m1_from (float): Start position of the first device.
m1_to (float): End position of the first device.
motor2 (Device): Second device that should be scanned.
m2_from (float): Start position of the second device.
m2_to (float): End position of the second device.
steps (int): Number of steps.
exp_time (float): Exposure time
Returns:
ScanReport: Status object.
Examples:
>>> a2scan(dev.motor1, -5, 5, dev.motor2, -8, 8, 10, 0.1)
"""
return scans.line_scan(
motor1,
m1_from,
m1_to,
motor2,
m2_from,
m2_to,
steps=steps,
exp_time=exp_time,
relative=False,
**kwargs
)
def dmesh(motor1, m1_from, m1_to, m1_steps, motor2, m2_from, m2_to, m2_steps, exp_time, **kwargs):
"""Relative mesh scan (grid scan) with two devices.
Args:
motor1 (Device): First device that should be scanned.
m1_from (float): Start position of the first device relative to its current position.
m1_to (float): End position of the first device relative to its current position.
m1_steps (int): Number of steps for motor1.
motor2 (Device): Second device that should be scanned.
m2_from (float): Start position of the second device relative to its current position.
m2_to (float): End position of the second device relative to its current position.
m2_steps (int): Number of steps for motor2.
exp_time (float): Exposure time
Returns:
ScanReport: Status object.
Examples:
>>> dmesh(dev.motor1, -5, 5, 10, dev.motor2, -8, 8, 10, 0.1)
"""
return scans.grid_scan(
motor1,
m1_from,
m1_to,
m1_steps,
motor2,
m2_from,
m2_to,
m2_steps,
exp_time=exp_time,
relative=True,
)
def amesh(motor1, m1_from, m1_to, m1_steps, motor2, m2_from, m2_to, m2_steps, exp_time, **kwargs):
"""Absolute mesh scan (grid scan) with two devices.
Args:
motor1 (Device): First device that should be scanned.
m1_from (float): Start position of the first device.
m1_to (float): End position of the first device.
m1_steps (int): Number of steps for motor1.
motor2 (Device): Second device that should be scanned.
m2_from (float): Start position of the second device.
m2_to (float): End position of the second device.
m2_steps (int): Number of steps for motor2.
exp_time (float): Exposure time
Returns:
ScanReport: Status object.
Examples:
>>> amesh(dev.motor1, -5, 5, 10, dev.motor2, -8, 8, 10, 0.1)
"""
return scans.grid_scan(
motor1,
m1_from,
m1_to,
m1_steps,
motor2,
m2_from,
m2_to,
m2_steps,
exp_time=exp_time,
relative=False,
)
def umv(*args) -> ScanReport:
"""Updated absolute move (i.e. blocking) for one or more devices.
Returns:
ScanReport: Status object.
Examples:
>>> umv(dev.samx, 1)
>>> umv(dev.samx, 1, dev.samy, 2)
"""
return scans.umv(*args, relative=False)
def umvr(*args) -> ScanReport:
"""Updated relative move (i.e. blocking) for one or more devices.
Returns:
ScanReport: Status object.
Examples:
>>> umvr(dev.samx, 1)
>>> umvr(dev.samx, 1, dev.samy, 2)
"""
return scans.umv(*args, relative=True)
def mv(*args) -> ScanReport:
"""Absolute move for one or more devices.
Returns:
ScanReport: Status object.
Examples:
>>> mv(dev.samx, 1)
>>> mv(dev.samx, 1, dev.samy, 2)
"""
return scans.mv(*args, relative=False)
def mvr(*args) -> ScanReport:
"""Relative move for one or more devices.
Returns:
ScanReport: Status object.
Examples:
>>> mvr(dev.samx, 1)
>>> mvr(dev.samx, 1, dev.samy, 2)
"""
return scans.mv(*args, relative=True)

42
bec_plugins/bec_client/startup/post_startup.py
View File

@@ -1,42 +0,0 @@
"""
Post startup script for the BEC client. This script is executed after the
IPython shell is started. It is used to load the beamline specific
information and to setup the prompts.
The script is executed in the global namespace of the IPython shell. This
means that all variables defined here are available in the shell.
If needed, bec command-line arguments can be parsed here. For example, to
parse the --session argument, add the following lines to the script:
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--session", help="Session name", type=str, default="my_default_session")
args = parser.parse_args()
if args.session == "my_session":
print("Loading my_session session")
from bec_plugins.bec_client.plugins.my_session import *
else:
print("Loading default session")
from bec_plugins.bec_client.plugins.default_session import *
"""
# pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module
from bec_lib import bec_logger
logger = bec_logger.logger
logger.info("Using the XTreme startup script.")
# SETUP BEAMLINE INFO
from bec_client.plugins.SLS.sls_info import OperatorInfo, SLSInfo
bec._beamline_mixin._bl_info_register(SLSInfo)
bec._beamline_mixin._bl_info_register(OperatorInfo)
# SETUP PROMPTS
bec._ip.prompts.username = "XTreme"
bec._ip.prompts.status = 1

24
bec_plugins/bec_client/startup/pre_startup.py
View File

@@ -1,24 +0,0 @@
"""
Pre-startup script for BEC client. This script is executed before the BEC client
is started. It can be used to set up the BEC client configuration. The script is
executed in the global namespace of the BEC client. This means that all
variables defined here are available in the BEC client.
To set up the BEC client configuration, use the ServiceConfig class. For example,
to set the configuration file path, add the following lines to the script:
import pathlib
from bec_lib import ServiceConfig
current_path = pathlib.Path(__file__).parent.resolve()
CONFIG_PATH = f"{current_path}/<path_to_my_config_file.yaml>"
config = ServiceConfig(CONFIG_PATH)
If this startup script defined a ServiceConfig object, the BEC client will use
it to configure itself. Otherwise, the BEC client will use the default config.
"""
from bec_lib import ServiceConfig
CONFIG_PATH = "/sls/X07MA/data/x07maop/SW/bec/bec/bec_client_config.yaml"
config = ServiceConfig(CONFIG_PATH)

175
bec_plugins/scan_server/scan_plugins/otf_scan.py
View File

@@ -1,175 +0,0 @@
import time
import numpy as np
from bec_lib import bec_logger
from scan_server.scans import FlyScanBase, ScanArgType, ScanBase
logger = bec_logger.logger
class OTFScan(FlyScanBase):
scan_name = "otf_scan"
scan_report_hint = "table"
required_kwargs = ["e1", "e2", "time"]
arg_input = []
arg_bundle_size = len(arg_input)
def __init__(self, *args, parameter=None, **kwargs):
"""Scans the energy from e1 to e2 in <time> minutes.
Examples:
>>> scans.otf_scan(e1=700, e2=740, time=4)
"""
super().__init__(parameter=parameter, **kwargs)
self.axis = []
self.scan_motors = []
self.num_pos = 0
self.mono = self.caller_kwargs.get("mono", "mono")
self.otf_device = self.caller_kwargs.get("otf", "otf")
def scan_core(self):
yield from self.stubs.set(
device=self.mono, value=self.caller_kwargs["e1"], wait_group="flyer"
)
yield from self.stubs.wait(device=[self.mono], wait_group="flyer", wait_type="move")
yield from self.stubs.kickoff(
device=self.otf_device,
parameter={
key: val for key, val in self.caller_kwargs.items() if key in ["e1", "e2", "time"]
},
)
yield from self.stubs.wait(device=[self.otf_device], wait_group="kickoff", wait_type="move")
yield from self.stubs.complete(device=self.otf_device)
target_diid = self.DIID - 1
while True:
yield from self.stubs.read_and_wait(group="primary", wait_group="readout_primary")
status = self.stubs.get_req_status(
device=self.otf_device, RID=self.metadata["RID"], DIID=target_diid
)
progress = self.stubs.get_device_progress(
device=self.otf_device, RID=self.metadata["RID"]
)
if progress:
self.num_pos = progress
if status:
break
time.sleep(1)
class HystScan(ScanBase):
scan_name = "hyst_scan"
scan_report_hint = "table"
required_kwargs = []
arg_input = [
ScanArgType.DEVICE,
ScanArgType.FLOAT,
ScanArgType.FLOAT,
ScanArgType.DEVICE,
ScanArgType.FLOAT,
ScanArgType.FLOAT,
]
arg_bundle_size = 3
scan_type = "step"
default_ramp_rate = 2
def __init__(self, *args, parameter=None, **kwargs):
"""
A hysteresis scan.
scans.hyst_scan(field_motor, start_field, end_field, mono, energy1, energy2)
Examples:
>>> scans.hyst_scan(dev.field_x, 0, 0.5, dev.mono, 600, 640, ramp_rate=2)
"""
super().__init__(parameter=parameter, **kwargs)
self.axis = []
self.flyer = list(self.caller_args.keys())[0]
self.energy_motor = list(self.caller_args.keys())[1]
self.scan_motors = [self.energy_motor, self.flyer]
self.flyer_positions = self.caller_args[self.flyer]
self._current_scan_motor_index = 0
self._scan_motor_direction = 1
self.ramp_rate = self.caller_kwargs.get("ramp_rate", self.default_ramp_rate)
def _calculate_positions(self) -> None:
self.positions = [[pos] for pos in self.caller_args[self.energy_motor]]
def prepare_positions(self):
self._calculate_positions()
self.num_pos = 0
yield None
self._check_limits()
def _at_each_point(self):
yield from self.stubs.read(group="primary", wait_group="primary", pointID=self.pointID)
self.pointID += 1
def _get_next_scan_motor_position(self):
while True:
yield self.positions[self._current_scan_motor_index][0]
if len(self.positions) - 1 == self._current_scan_motor_index or (
self._current_scan_motor_index == 0 and self._scan_motor_direction < 0
):
self._scan_motor_direction *= -1
self._current_scan_motor_index += self._scan_motor_direction
def scan_core(self):
# yield from self._move_and_wait(self.positions[0])
status = yield from self.stubs.send_rpc_and_wait(
"field_x", "ramprate.set", self.default_ramp_rate
)
status.wait()
yield from self.stubs.set(
device=self.flyer, value=self.flyer_positions[0], wait_group="flyer"
)
yield from self.stubs.wait(device=[self.flyer], wait_group="flyer", wait_type="move")
status = yield from self.stubs.send_rpc_and_wait("field_x", "ramprate.set", self.ramp_rate)
status.wait()
# send the slow motor on its way
yield from self.stubs.set(
device=self.flyer,
value=self.flyer_positions[1],
wait_group="flyer",
)
flyer_done = False
pos_generator = self._get_next_scan_motor_position()
target_DIID = self.DIID - 1
dev = self.device_manager.devices
while not flyer_done:
flyer_done = bool(
self.stubs.get_req_status(
device=self.flyer, DIID=target_DIID, RID=self.metadata["RID"]
)
)
val = next(pos_generator)
logger.info(f"Moving mono to {val}.")
yield from self.stubs.set(device=self.energy_motor, value=val, wait_group="mono")
yield from self.stubs.wait(
device=[self.energy_motor], wait_group="mono", wait_type="move"
)
monitored_devices = [dev.name for dev in dev.monitored_devices([])]
yield from self.stubs.read_and_wait(
device=[self.flyer, self.scan_motors[0], *monitored_devices],
wait_group="readout_primary",
pointID=self.pointID,
)
# time.sleep(1)
self.pointID += 1
self.scan_motors[0]
self.num_pos += 1
status = yield from self.stubs.send_rpc_and_wait(
"field_x", "ramprate.set", self.default_ramp_rate
)
status.wait()
def return_to_start(self):
yield None

11
deployment/autodeploy_versions
View File

@@ -1,11 +0,0 @@
# This file is used to select the BEC and Ophyd Devices version for the auto deployment process.
# Do not edit this file unless you know what you are doing!
# The version can be a git tag, branch or commit hash.
# BEC version to use
BEC_AUTODEPLOY_VERSION="master"
# ophyd_devices version to use
OPHYD_DEVICES_AUTODEPLOY_VERSION="master"

18
deployment/bec-server-config.yaml
View File

@@ -1,18 +0,0 @@
redis:
host: localhost
port: 6379
mongodb:
host: localhost
port: 27017
scibec:
host: http://[::1]
port: 3030
beamline: "XTREME"
service_config:
general:
reset_queue_on_cancel: True
enforce_ACLs: False
file_writer:
plugin: default_NeXus_format
base_path: ./

27
deployment/deploy.sh
View File

@@ -1,27 +0,0 @@
# deployment script to be translated to Ansible
# NOT NEEDED since the beamline repo will be autodeployed
# BEAMLINE_REPO=gitlab.psi.ch:bec/csaxs-bec.git
# git clone git@$BEAMLINE_REPO
module add psi-python311/2024.02
# start redis
docker run --network=host --name redis-bec -d redis
# alternative:
# conda install -y redis; redis-server &
# get the target versions for ophyd_devices and BEC
source ./csaxs-bec/deployment/autodeploy_versions
git clone -b $OPHYD_DEVICES_AUTODEPLOY_VERSION https://gitlab.psi.ch/bec/ophyd_devices.git
git clone -b $BEC_AUTODEPLOY_VERSION https://gitlab.psi.ch/bec/bec.git
# install BEC
cd bec
source ./bin/install_bec_dev.sh
cd ../
# start the BEC server
bec-server start --config ./xtreme-bec/deployment/bec-server-config.yaml

74
pyproject.toml Normal file
View File

@@ -0,0 +1,74 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "xtreme_bec"
version = "0.0.0"
description = "Custom device implementations based on the ophyd hardware abstraction layer"
requires-python = ">=3.10"
classifiers = [
"Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3",
"Topic :: Scientific/Engineering",
]
dependencies = []
[project.optional-dependencies]
dev = [
"black",
"isort",
"coverage",
"pylint",
"pytest",
"pytest-random-order",
"ophyd_devices",
"bec_server",
]
[project.entry-points."bec"]
plugin_bec = "xtreme_bec"
[project.entry-points."bec.deployment.device_server"]
plugin_ds_startup = "xtreme_bec.deployments.device_server.startup:run"
[project.entry-points."bec.file_writer"]
plugin_file_writer = "xtreme_bec.file_writer"
[project.entry-points."bec.scans"]
plugin_scans = "xtreme_bec.scans"
[project.entry-points."bec.ipython_client_startup"]
plugin_ipython_client_pre = "xtreme_bec.bec_ipython_client.startup.pre_startup"
plugin_ipython_client_post = "xtreme_bec.bec_ipython_client.startup"
[project.entry-points."bec.widgets.auto_updates"]
plugin_widgets_update = "xtreme_bec.bec_widgets.auto_updates:PlotUpdate"
[project.entry-points."bec.widgets.user_widgets"]
plugin_widgets = "xtreme_bec.bec_widgets.widgets"
[tool.hatch.build.targets.wheel]
include = ["*"]
[tool.isort]
profile = "black"
line_length = 100
multi_line_output = 3
include_trailing_comma = true
[tool.black]
line-length = 100
skip-magic-trailing-comma = true
[tool.pylint.basic]
# Good variable names regexes, separated by a comma. If names match any regex,
# they will always be accepted
good-names-rgxs = [
".*scanID.*",
".*RID.*",
".*pointID.*",
".*ID.*",
".*_2D.*",
".*_1D.*",
]

21
setup.cfg
View File

@@ -1,21 +0,0 @@
[metadata]
name = bec_plugins
description = BEC plugins to modify the behaviour of services within the BEC framework
long_description = file: README.md
long_description_content_type = text/markdown
url = https://gitlab.psi.ch/bec/bec
project_urls =
Bug Tracker = https://gitlab.psi.ch/bec/bec/issues
classifiers =
Programming Language :: Python :: 3
Development Status :: 3 - Alpha
Topic :: Scientific/Engineering
[options]
package_dir =
= .
packages = find:
python_requires = >=3.10
[options.packages.find]
where = .

7
setup.py
View File

@@ -1,7 +0,0 @@
from setuptools import setup
if __name__ == "__main__":
setup(
install_requires=[],
extras_require={"dev": ["pytest", "pytest-random-order", "coverage"]},
)

31
tests/tests_bec_ipython_client/README.md Normal file
View File

@@ -0,0 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

31
tests/tests_bec_widgets/README.md Normal file
View File

@@ -0,0 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

31
tests/tests_devices/README.md Normal file
View File

@@ -0,0 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

31
tests/tests_file_writer/README.md Normal file
View File

@@ -0,0 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

31
tests/tests_scans/README.md Normal file
View File

@@ -0,0 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

0
bec_plugins/bec_client/hli/__init__.py → xtreme_bec/__init__.py
View File

0
bec_plugins/bec_client/plugins/__init__.py → xtreme_bec/bec_ipython_client/__init__.py
View File

0
bec_plugins/bec_client/startup/__init__.py → xtreme_bec/bec_ipython_client/high_level_interface/__init__.py
View File

0
bec_plugins/device_server/__init__.py → xtreme_bec/bec_ipython_client/plugins/__init__.py
View File

0
bec_plugins/scan_server/__init__.py → xtreme_bec/bec_ipython_client/startup/__init__.py
View File

36
xtreme_bec/bec_ipython_client/startup/post_startup.py Normal file
View File

@@ -0,0 +1,36 @@
"""
Post startup script for the BEC client. This script is executed after the
IPython shell is started. It is used to load the beamline specific
information and to setup the prompts.
The script is executed in the global namespace of the IPython shell. This
means that all variables defined here are available in the shell.
While command-line arguments have to be set in the pre-startup script, the
post-startup script can be used to load beamline specific information and
to setup the prompts.
from bec_lib import bec_logger
logger = bec_logger.logger
# pylint: disable=import-error
_args = _main_dict["args"]
_session_name = "cSAXS"
if _args.session.lower() == "lamni":
from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
from csaxs_bec.bec_ipython_client.plugins.LamNI import *
_session_name = "LamNI"
lamni = LamNI(bec)
logger.success("LamNI session loaded.")
elif _args.session.lower() == "csaxs":
print("Loading cSAXS session")
from csaxs_bec.bec_ipython_client.plugins.cSAXS import *
logger.success("cSAXS session loaded.")
"""
# pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module

23
xtreme_bec/bec_ipython_client/startup/pre_startup.py Normal file
View File

@@ -0,0 +1,23 @@
"""
Pre-startup script for BEC client. This script is executed before the BEC client
is started. It can be used to add additional command line arguments.
"""
from bec_lib.service_config import ServiceConfig
def extend_command_line_args(parser):
"""
Extend the command line arguments of the BEC client.
"""
# parser.add_argument("--session", help="Session name", type=str, default="cSAXS")
return parser
# def get_config() -> ServiceConfig:
# """
# Create and return the service configuration.
# """
# return ServiceConfig(redis={"host": "localhost", "port": 6379})

0
xtreme_bec/bec_widgets/__init__.py Normal file
View File

0
xtreme_bec/bec_widgets/widgets/__init__.py Normal file
View File

0
xtreme_bec/deployments/__init__.py Normal file
View File

0
xtreme_bec/deployments/device_server/__init__.py Normal file
View File

0
bec_plugins/device_server/startup.py → xtreme_bec/deployments/device_server/startup.py
View File

0
xtreme_bec/device_configs/__init__.py Normal file
View File

0
xtreme_bec/devices/__init__.py Normal file
View File

0
xtreme_bec/file_writer/__init__.py Normal file
View File

0
xtreme_bec/scans/__init__.py Normal file
View File

32
xtreme_bec/scans/scan_plugin_template.py Normal file
View File

@@ -0,0 +1,32 @@
"""
SCAN PLUGINS
All new scans should be derived from ScanBase. ScanBase provides various methods that can be customized and overriden
but they are executed in a specific order:
- self.initialize # initialize the class if needed
- self.read_scan_motors # used to retrieve the start position (and the relative position shift if needed)
- self.prepare_positions # prepare the positions for the scan. The preparation is split into multiple sub fuctions:
- self._calculate_positions # calculate the positions
- self._set_positions_offset # apply the previously retrieved scan position shift (if needed)
- self._check_limits # tests to ensure the limits won't be reached
- self.open_scan # send an open_scan message including the scan name, the number of points and the scan motor names
- self.stage # stage all devices for the upcoming acquisiton
- self.run_baseline_readings # read all devices to get a baseline for the upcoming scan
- self.pre_scan # perform additional actions before the scan starts
- self.scan_core # run a loop over all position
- self._at_each_point(ind, pos) # called at each position with the current index and the target positions as arguments
- self.finalize # clean up the scan, e.g. move back to the start position; wait everything to finish
- self.unstage # unstage all devices that have been staged before
- self.cleanup # send a close scan message and perform additional cleanups if needed
"""
# import time
# import numpy as np
# from bec_lib import MessageEndpoints, bec_logger, messages
# from bec_server.scan_server.errors import ScanAbortion
# from bec_server.scan_server.scans import FlyScanBase, RequestBase, ScanArgType, ScanBase
# logger = bec_logger.logger

0
xtreme_bec/services/__init__.py Normal file
View File