bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor: upgraded to new plugin structure

Browse Source
This commit is contained in:
wakonig_k
2024-04-24 10:01:23 +02:00
parent e3515e6435
commit 71ebbd86be
42 changed files with 408 additions and 396 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

31
.gitlab-ci.yml
View File

@@ -3,6 +3,18 @@
# https://hub.docker.com/r/library/python/tags/
image: $CI_DEPENDENCY_PROXY_GROUP_IMAGE_PREFIX/python:3.10
workflow:
rules:
- if: $CI_PIPELINE_SOURCE == "schedule"
- if: $CI_PIPELINE_SOURCE == "web"
- if: $CI_PIPELINE_SOURCE == "merge_request_event"
- if: $CI_COMMIT_BRANCH && $CI_OPEN_MERGE_REQUESTS
when: never
- if: $CI_COMMIT_BRANCH
include:
- template: Security/Secret-Detection.gitlab-ci.yml
#commands to run in the Docker container before starting each job.
before_script:
- pip install -e .[dev]
@@ -10,10 +22,14 @@ before_script:
# different stages in the pipeline
stages:
- Formatter
- Test
- test
- AdditionalTests
- Deploy
secret_detection:
before_script:
- ''
formatter:
stage: Formatter
script:
@@ -21,7 +37,7 @@ formatter:
- black --check --diff --color --line-length=100 ./
pytest:
stage: Test
stage: test
script:
- pytest -v --random-order ./tests
@@ -38,3 +54,14 @@ tests-3.12:
stage: AdditionalTests
image: $CI_DEPENDENCY_PROXY_GROUP_IMAGE_PREFIX/python:3.12
allow_failure: true
config_test:
stage: test
script:
- ophyd_test --config ./debye_bec/device_configs/ --output ./config_tests
artifacts:
paths:
- ./config_tests
when: on_failure
expire_in: "30 days"
allow_failure: false

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)

1
bec_plugins/bec_client/plugins/__init__.py
View File

@@ -1 +0,0 @@

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

@@ -1,46 +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
import argparse
from bec_lib import bec_logger
logger = bec_logger.logger
logger.info("Using the Debye startup script.")
parser = argparse.ArgumentParser()
parser.add_argument("--session", help="Session name", type=str, default="Debye")
args = parser.parse_args()
# 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 = "Debye"
bec._ip.prompts.status = 1

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

@@ -1,25 +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.
"""
# example:
# current_path = pathlib.Path(__file__).parent.resolve()
# CONFIG_PATH = f"{current_path}/../../../bec_config.yaml"
# config = ServiceConfig(CONFIG_PATH)

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

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

0
bec_plugins/scan_server/__init__.py → debye_bec/bec_ipython_client/high_level_interface/__init__.py
View File

0
debye_bec/bec_ipython_client/plugins/__init__.py Normal file
View File

0
debye_bec/bec_ipython_client/startup/__init__.py Normal file
View File

36
debye_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

14
debye_bec/bec_ipython_client/startup/pre_startup.py Normal file
View File

@@ -0,0 +1,14 @@
"""
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.
"""
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

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

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

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

11
debye_bec/deployments/device_server/startup.py Normal file
View File

@@ -0,0 +1,11 @@
import os
def setup_epics_ca():
# os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "NO"
# os.environ["EPICS_CA_ADDR_LIST"] = "129.129.122.255 sls-x12sa-cagw.psi.ch:5836"
os.environ["PYTHONIOENCODING"] = "latin1"
def run():
setup_epics_ca()

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

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

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

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

32
debye_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

4
bec_plugins/services/NIDAQ_writer/NIDAQ_writer.py → debye_bec/services/NIDAQ_writer/NIDAQ_writer.py
View File

@@ -9,7 +9,7 @@ import h5py
import numpy as np
from bec_lib import MessageEndpoints, RedisConnector, ServiceConfig, bec_logger, messages
from bec_lib.bec_service import BECService
from bec_lib.file_utils import FileWriterMixin
from bec_lib.file_utils import FileWriter
from bec_lib.redis_connector import MessageObject
logger = bec_logger.logger
@@ -27,7 +27,7 @@ class NIDAQWriterService(BECService):
super().__init__(config=config, connector_cls=connector_cls, unique_service=True)
self.queue = queue.Queue()
config = self._service_config.service_config.get("file_writer")
self.writer_mixin = FileWriterMixin(config)
self.writer_mixin = FileWriter(service_config=config)
self._scan_status_consumer = None
self._ni_data_consumer = None
self._ni_data_event = None

0
bec_plugins/services/NIDAQ_writer/__init__.py → debye_bec/services/NIDAQ_writer/__init__.py
View File

14
bec_plugins/services/NIDAQ_writer/scan_status.py → debye_bec/services/NIDAQ_writer/scan_status.py
View File

@@ -24,19 +24,9 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Scan status helper")
command = parser.add_subparsers(dest="command")
start = command.add_parser("start", help="Start a new scan")
start.add_argument(
"--scan_number",
type=int,
required=True,
help="Scan number",
)
start.add_argument("--scan_number", type=int, required=True, help="Scan number")
stop = command.add_parser("stop", help="Stop the scan")
stop.add_argument(
"--scan_number",
type=int,
required=True,
help="Scan number",
)
stop.add_argument("--scan_number", type=int, required=True, help="Scan number")
args = parser.parse_args()
send_scan_status(args.scan_number, args.command)

10
bec_plugins/services/NIDAQ_writer/utils/nidaq_sim.py → debye_bec/services/NIDAQ_writer/utils/nidaq_sim.py
View File

@@ -2,7 +2,7 @@ import threading
import time
import numpy as np
from bec_lib import MessageEndpoints, RedisConnector, ServiceConfig, bec_logger, messages
from bec_lib import RedisConnector, messages
class NIDAQSim(threading.Thread):
@@ -13,10 +13,7 @@ class NIDAQSim(threading.Thread):
index = 0
producer = RedisConnector(["localhost:6379"]).producer()
signal = np.asarray(range(index, index + 600000))
signals = {
"signal1": signal,
"signal2": signal,
}
signals = {"signal1": signal, "signal2": signal}
msg = messages.DeviceMessage(signals=signals)
msg = msg.dumps()
@@ -40,9 +37,6 @@ class NIDAQSim(threading.Thread):
time.sleep(0.5)
print(f"Elapsed time: {time.time() - start}")
print(f"Total time: {time.time() - total_time}")
print(f"FPS: {index / (time.time() - total_time)}")
print(f"Signal size: {signal.nbytes/1e6*2} MB")
if __name__ == "__main__":

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

14
bec_plugins/services/launch_writer.py → debye_bec/services/launch_writer.py
View File

@@ -2,14 +2,11 @@ import argparse
import threading
from bec_lib import RedisConnector, ServiceConfig, bec_logger
from NIDAQ_writer import NIDAQWriterService
from debye_bec.services.NIDAQ_writer import NIDAQWriterService
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument(
"--config",
default="",
help="path to the config file",
)
parser.add_argument("--config", default="", help="path to the config file")
clargs = parser.parse_args()
config_path = clargs.config
@@ -17,10 +14,7 @@ config = ServiceConfig(config_path)
bec_logger.level = bec_logger.LOGLEVEL.INFO
logger = bec_logger.logger
bec_server = NIDAQWriterService(
config=config,
connector_cls=RedisConnector,
)
bec_server = NIDAQWriterService(config=config, connector_cls=RedisConnector)
try:
event = threading.Event()
# pylint: disable=E1102

70
pyproject.toml Normal file
View File

@@ -0,0 +1,70 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "debye_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 = ["numpy", "bec_lib", "h5py", "ophyd_devices"]
[project.optional-dependencies]
dev = [
"bec_server",
"black",
"isort",
"coverage",
"pylint",
"pytest",
"pytest-random-order",
]
[project.entry-points."bec"]
plugin_bec = "debye_bec"
[project.entry-points."bec.deployment.device_server"]
plugin_ds_startup = "debye_bec.deployments.device_server.startup:run"
[project.entry-points."bec.file_writer"]
plugin_file_writer = "debye_bec.file_writer"
[project.entry-points."bec.scans"]
plugin_scans = "debye_bec.scans"
[project.entry-points."bec.ipython_client_startup"]
plugin_ipython_client_pre = "debye_bec.bec_ipython_client.startup.pre_startup"
plugin_ipython_client_post = "debye_bec.bec_ipython_client.startup"
[project.entry-points."bec.widgets"]
plugin_widgets = "debye_bec.bec_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=["numpy", "h5py", "bec-lib"],
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).

31
tests/tests_services/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).

26
tests/test_file_writer_service.py → tests/tests_services/test_file_writer_service.py
View File

@@ -5,7 +5,7 @@ import pytest
from bec_lib import MessageEndpoints, ServiceConfig, messages
from bec_lib.redis_connector import MessageObject
from bec_plugins.services.NIDAQ_writer import NIDAQWriterService
from debye_bec.services.NIDAQ_writer import NIDAQWriterService
def test_nidaq_starts_consumers():
@@ -32,8 +32,7 @@ class NIWriterMock(NIDAQWriterService):
@pytest.fixture(scope="function")
def nidaq():
service = NIWriterMock(
config=ServiceConfig(redis={"host": "test", "port": 6379}),
connector_cls=mock.MagicMock(),
config=ServiceConfig(redis={"host": "test", "port": 6379}), connector_cls=mock.MagicMock()
)
yield service
@@ -46,11 +45,8 @@ def test_nidaq_scan_status_consumer(nidaq):
def test_scan_status_callback(nidaq):
scan_status_msg = messages.ScanStatusMessage(scanID="test", status="open", info={})
msg_obj = MessageObject(
topic="test",
value=scan_status_msg.dumps(),
)
scan_status_msg = messages.ScanStatusMessage(scan_id="test", status="open", info={})
msg_obj = MessageObject(topic="test", value=scan_status_msg.dumps())
with mock.patch.object(nidaq, "handle_scan_status") as mock_handle:
nidaq._scan_status_callback(msg_obj, nidaq)
mock_handle.assert_called_once_with(scan_status_msg)
@@ -83,10 +79,10 @@ def test_nidaq_reads_data_from_strea(nidaq):
mock_handle.assert_called_once()
@pytest.mark.parametrize("scan_status", ["open", "closed", "aborted", "halted", None])
@pytest.mark.parametrize("scan_status", ["open", "closed", "aborted", "halted"])
def test_nidaq_handle_scan_status(nidaq, scan_status):
scan_status_msg = messages.ScanStatusMessage(
scanID="test", status=scan_status, info={"scan_number": 5}
scan_id="test", status=scan_status, info={"scan_number": 5}
)
nidaq.handle_scan_status(scan_status_msg)
if scan_status == "open":
@@ -114,7 +110,7 @@ def test_nidaq_handle_ni_data(nidaq):
def test_nidaq_write_data_without_filename(nidaq):
signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
nidaq.write_data(signal)
mock_h5py.File.assert_not_called()
@@ -122,7 +118,7 @@ def test_nidaq_write_data_without_filename(nidaq):
def test_nidaq_write_data_with_filename(nidaq):
signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
nidaq.filename = "test.h5"
with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
nidaq.write_data(signal)
mock_h5py.File.assert_called_once_with("test.h5", "a")
@@ -131,7 +127,7 @@ def test_nidaq_write_data_reshape(nidaq):
signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
nidaq.filename = "test.h5"
nidaq.reshape_dataset = True
with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
nidaq.write_data(signal)
mock_h5py.File.assert_called_once_with("test.h5", "a")
@@ -140,7 +136,7 @@ def test_nidaq_write_data_without_reshape(nidaq):
signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
nidaq.filename = "test.h5"
nidaq.reshape_dataset = False
with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
nidaq.write_data(signal)
mock_h5py.File.assert_called_once_with("test.h5", "a")
file_handle = mock_h5py.File().__enter__()
@@ -161,7 +157,7 @@ def test_nidaq_write_data_reshapes_data(nidaq):
signal = {"signal1": np.asarray(range(20)), "signal2": np.asarray(range(20))}
nidaq.filename = "test.h5"
nidaq.reshape_dataset = True
with mock.patch("bec_plugins.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
with mock.patch("debye_bec.services.NIDAQ_writer.NIDAQ_writer.h5py") as mock_h5py:
file_handle = mock_h5py.File().__enter__()
file_handle.__contains__.side_effect = signal.__contains__
nidaq.write_data(signal)