.copier-answers.yml

@@ -2,7 +2,7 @@
 # It is needed to track the repo template version, and editing may break things.
 # This file will be overwritten by copier on template updates.
 
-_commit: v1.4.1
+_commit: v1.5.1
 _src_path: https://github.com/bec-project/plugin_copier_template.git
 make_commit: false
 project_name: sim_bec

.gitea/workflows/ci.yml

@@ -95,8 +95,19 @@ jobs:
           uv pip install --system -e ./bec/bec_ipython_client
           uv pip install --system -e ./bec/bec_server[dev]
           uv pip install --system -e ./bec_widgets[dev,pyside6]
-          uv pip install --system -e ./sim_bec
+          uv pip install --system -e ./sim_bec[dev]
 
-      - name: Run Pytest with Coverage
+      - name: Run Tests with Coverage
         id: coverage
-        run: pytest --random-order --cov=./sim_bec --cov-config=./sim_bec/pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail ./sim_bec/tests/ || test $? -eq 5
+        shell: bash
+        run: |
+          set +e
+          coverage run --branch --source=./sim_bec -m pytest --random-order ./sim_bec/tests/
+          status=$?
+          # Allow pytest exit code 5 so repositories without tests do not fail CI.
+          if [ "$status" -ne 0 ] && [ "$status" -ne 5 ]; then
+            exit "$status"
+          fi
+          if [ "$status" -eq 0 ]; then
+            coverage report
+          fi

pyproject.toml

@@ -36,6 +36,9 @@ plugin_ds_startup = "sim_bec.deployments.device_server.startup:run"
 [project.entry-points."bec.file_writer"]
 plugin_file_writer = "sim_bec.file_writer"
 
+[project.entry-points."bec.file_writer.storage_copy"]
+plugin_storage_copy = "sim_bec.file_writer.storage_copy:plugin_storage_copy"
+
 [project.entry-points."bec.scans"]
 plugin_scans = "sim_bec.scans"
 

sim_bec/file_writer/storage_copy.py

@@ -0,0 +1,76 @@
+"""
+Beamline storage-copy hook for sim_bec.
+
+Use this module to handle ``bec.beamline_storage_copy(...)`` requests with
+beamline-specific file transfer logic.
+
+The plugin_storage_copy function is commented out by default. Uncomment and implement it to handle
+beamline-specific storage copy operations. The function should accept the source file path,
+a scope identifier, the active BEC account, and an optional subdirectory for the destination. It should
+perform the necessary checks and copy the file to the appropriate location based on the provided scope.
+The subdirectory, if provided, has already been sanitized to ensure it is safe for use in file paths.
+"""
+
+# import os
+# import shutil
+
+# from bec_lib.logger import bec_logger
+
+# logger = bec_logger.logger
+
+
+# def plugin_storage_copy(
+#     source_file: str, scope: str, active_account: str, subdir: str | None = None
+# ) -> None:
+#     """
+#     Run a beamline-defined storage copy operation.
+
+#     Args:
+#         source_file: Source file that should be copied.
+#         scope: Beamline-defined copy scope identifier.
+#         active_account: Active BEC account, if one is available.
+#         subdir: Optional sanitized relative destination subdirectory.
+#     """
+#     supported_file_extensions = [
+#         ".h5",
+#         ".txt",
+#         ".csv",
+#         ".log",
+#         ".json",
+#         ".png",
+#         ".jpg",
+#         ".jpeg",
+#         ".tiff",
+#         ".yaml",
+#         ".yml",
+#     ]
+#     if not any(source_file.endswith(ext) for ext in supported_file_extensions):
+#         logger.error(
+#             f"Unsupported file type for '{source_file}'. Supported extensions: "
+#             f"{supported_file_extensions}"
+#         )
+#         return
+
+#     # Example storage mapping. Replace this with beamline-specific scopes and paths.
+#     storage_locations = {"alignment": f"/sls/x99sa/data/p12345/raw/bec/alignment/{active_account}/"}
+#     if scope not in storage_locations:
+#         logger.error(
+#             f"Received unknown scope '{scope}'. Available scopes: {list(storage_locations.keys())}"
+#         )
+#         return
+
+#     if not os.path.isfile(source_file):
+#         logger.error(f"File '{source_file}' does not exist or is not reachable.")
+#         return
+
+#     destination_dir = storage_locations[scope]
+#     if subdir:
+#         destination_dir = os.path.join(destination_dir, subdir)
+#     os.makedirs(destination_dir, exist_ok=True)
+#     destination_file = os.path.join(destination_dir, os.path.basename(source_file))
+
+#     try:
+#         shutil.copy(source_file, destination_file)
+#         logger.info(f"Copied '{source_file}' to '{destination_file}'")
+#     except Exception as exc:
+#         logger.error(f"Failed to copy '{source_file}' to '{destination_file}': {exc}")