1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-05-10 00:32:10 +02:00

Compare commits

..

22 Commits

Author SHA1 Message Date
wakonig_k 7e4e37ad16 fix(web_links): update documentation links in BECWebLinksMixin 2026-05-05 16:09:47 +02:00
semantic-release c1d5069a48 3.8.0
Automatically generated by python-semantic-release
2026-05-01 15:16:03 +00:00
wyzula_j 0b1f0b4c26 fix(dock_area): change to show_dialo=False for CLI profile baseline restore 2026-05-01 17:15:03 +02:00
wyzula_j cc825972c2 fix(dock_area): cli call load_profile has restore_baseline kwarg 2026-05-01 17:15:03 +02:00
wyzula_j 17865a2c33 feat(dock_area): add CLI restore current profile from baseline with optional confirmation dialog 2026-05-01 17:15:03 +02:00
semantic-release 0728811238 3.7.3
Automatically generated by python-semantic-release
2026-05-01 11:33:30 +00:00
wyzula_j 717d74b19e test(dock_area): remove low-value tests 2026-05-01 13:32:46 +02:00
wyzula_j dd32caf6e8 fix(dock_area): profile names changed, default->baseline, user->runtime 2026-05-01 13:32:46 +02:00
semantic-release 603edede9c 3.7.2
Automatically generated by python-semantic-release
2026-04-29 13:13:24 +00:00
copilot-swe-agent[bot] 30ef25533a fix(workspace-actions): use try/finally and restore previous blocked state in refresh_profiles
Agent-Logs-Url: https://github.com/bec-project/bec_widgets/sessions/004cb4bc-5015-485e-a803-1e63876b7024

Co-authored-by: wyzula-jan <133381102+wyzula-jan@users.noreply.github.com>
2026-04-29 15:12:35 +02:00
wyzula_j 73b44cffb2 fix(dock-area): avoid switching profile when saving new profile 2026-04-29 15:12:35 +02:00
wakonig_k a614d662d6 test: fix assertions after updating ophyd devices templates
Co-authored-by: Copilot <copilot@github.com>
2026-04-28 14:45:42 +02:00
wakonig_k 3f1aa80756 chore: update header comments in script files to indicate AI generation 2026-04-24 16:17:44 +02:00
wakonig_k 409c9e5bfa ci: increase threshold to 20 percent 2026-04-22 13:10:15 +02:00
wakonig_k 19b5c8f724 ci: fix benchmark upload 2026-04-22 13:10:15 +02:00
wyzula_j 5056ef8946 test: remove references to "scan_motors" in tests 2026-04-22 08:12:39 +02:00
wakonig_k 551d38d901 build: add pytest-benchmark dependency 2026-04-21 13:58:38 +02:00
wakonig_k 999b7a2321 ci: add benchmark workflow 2026-04-21 13:58:38 +02:00
semantic-release 5dc373bd8e 3.7.1
Automatically generated by python-semantic-release
2026-04-21 11:56:57 +00:00
appel_c 91afc775d5 test: fix exit status and status access in tests 2026-04-21 13:56:12 +02:00
appel_c 55694ff2b9 fix(heatmap): fix access to status from metadata 2026-04-21 13:56:12 +02:00
perl_d 5b68a51aaa tests: skip broken and mark with issue 2026-04-21 12:43:31 +02:00
32 changed files with 1915 additions and 487 deletions
+169
View File
@@ -0,0 +1,169 @@
##########################
### AI-generated file. ###
##########################
"""Aggregate and merge benchmark JSON files.
The workflow runs the same benchmark suite on multiple independent runners.
This script reads every JSON file produced by those attempts, normalizes the
contained benchmark values, and writes a compact mapping JSON where each value is
the median across attempts. It can also merge independent hyperfine JSON files
from one runner into a single hyperfine-style JSON file.
"""
from __future__ import annotations
import argparse
import json
import statistics
from pathlib import Path
from typing import Any
from compare_benchmarks import Benchmark, extract_benchmarks
def collect_benchmarks(paths: list[Path]) -> dict[str, list[Benchmark]]:
"""Collect benchmarks from multiple JSON files.
Args:
paths (list[Path]): Paths to hyperfine, pytest-benchmark, or compact
mapping JSON files.
Returns:
dict[str, list[Benchmark]]: Benchmarks grouped by benchmark name.
"""
collected: dict[str, list[Benchmark]] = {}
for path in paths:
for name, benchmark in extract_benchmarks(path).items():
collected.setdefault(name, []).append(benchmark)
return collected
def aggregate(collected: dict[str, list[Benchmark]]) -> dict[str, dict[str, object]]:
"""Aggregate grouped benchmarks using the median value.
Args:
collected (dict[str, list[Benchmark]]): Benchmarks grouped by benchmark
name.
Returns:
dict[str, dict[str, object]]: Compact mapping JSON data. Each benchmark
contains ``value``, ``unit``, ``metric``, ``attempts``, and
``attempt_values``.
"""
aggregated: dict[str, dict[str, object]] = {}
for name, benchmarks in sorted(collected.items()):
values = [benchmark.value for benchmark in benchmarks]
unit = next((benchmark.unit for benchmark in benchmarks if benchmark.unit), "")
metric = next((benchmark.metric for benchmark in benchmarks if benchmark.metric), "value")
aggregated[name] = {
"value": statistics.median(values),
"unit": unit,
"metric": f"median-of-attempt-{metric}",
"attempts": len(values),
"attempt_values": values,
}
return aggregated
def merge_hyperfine_results(paths: list[Path]) -> dict[str, Any]:
"""Merge hyperfine result files.
Args:
paths (list[Path]): Hyperfine JSON files to merge.
Returns:
dict[str, Any]: Hyperfine-style JSON object containing all result rows.
Raises:
ValueError: If any file has no hyperfine ``results`` list.
"""
merged: dict[str, Any] = {"results": []}
for path in paths:
data = json.loads(path.read_text(encoding="utf-8"))
results = data.get("results", []) if isinstance(data, dict) else None
if not isinstance(results, list):
raise ValueError(f"{path} has no hyperfine results list")
merged["results"].extend(results)
return merged
def main_from_paths(input_dir: Path, output: Path) -> int:
"""Aggregate all JSON files in a directory and write the result.
Args:
input_dir (Path): Directory containing benchmark JSON files.
output (Path): Path where the aggregate JSON should be written.
Returns:
int: Always ``0`` on success.
Raises:
ValueError: If no JSON files are found in ``input_dir``.
"""
paths = sorted(input_dir.rglob("*.json"))
if not paths:
raise ValueError(f"No benchmark JSON files found in {input_dir}")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(aggregate(collect_benchmarks(paths)), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return 0
def merge_from_paths(input_dir: Path, output: Path) -> int:
"""Merge all hyperfine JSON files in a directory and write the result.
Args:
input_dir (Path): Directory containing hyperfine JSON files.
output (Path): Path where the merged JSON should be written.
Returns:
int: Always ``0`` on success.
Raises:
ValueError: If no JSON files are found in ``input_dir``.
"""
paths = sorted(input_dir.glob("*.json"))
if not paths:
raise ValueError(f"No hyperfine JSON files found in {input_dir}")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(merge_hyperfine_results(paths), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return 0
def main() -> int:
"""Run the benchmark aggregation command line interface.
Returns:
int: Always ``0`` on success.
"""
parser = argparse.ArgumentParser()
parser.add_argument(
"--mode",
choices=("aggregate", "merge-hyperfine"),
default="aggregate",
help="Operation to perform.",
)
parser.add_argument("--input-dir", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
if args.mode == "merge-hyperfine":
return merge_from_paths(input_dir=args.input_dir, output=args.output)
return main_from_paths(input_dir=args.input_dir, output=args.output)
if __name__ == "__main__":
raise SystemExit(main())
+454
View File
@@ -0,0 +1,454 @@
##########################
### AI-generated file. ###
##########################
"""Compare benchmark JSON files and write a GitHub Actions summary.
The script supports JSON emitted by hyperfine, JSON emitted by pytest-benchmark,
and a compact mapping format generated by ``aggregate_benchmarks.py``. Timing
formats prefer median values and fall back to mean values when median values are
not present.
"""
from __future__ import annotations
import argparse
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass(frozen=True)
class Benchmark:
"""Normalized benchmark result.
Attributes:
name (str): Stable benchmark name used to match baseline and current results.
value (float): Numeric benchmark value used for comparison.
unit (str): Display unit for the value, for example ``"s"``.
metric (str): Source metric name, for example ``"median"`` or ``"mean"``.
"""
name: str
value: float
unit: str
metric: str = "value"
@dataclass(frozen=True)
class Comparison:
"""Comparison between one baseline benchmark and one current benchmark.
Attributes:
name (str): Benchmark name.
baseline (float): Baseline benchmark value.
current (float): Current benchmark value.
delta_percent (float): Percent change from baseline to current.
unit (str): Display unit for both values.
metric (str): Current result metric used for comparison.
regressed (bool): Whether the change exceeds the configured threshold in
the worse direction.
improved (bool): Whether the change exceeds the configured threshold in
the better direction.
"""
name: str
baseline: float
current: float
delta_percent: float
unit: str
metric: str
regressed: bool
improved: bool
def _read_json(path: Path) -> Any:
"""Read JSON data from a file.
Args:
path (Path): Path to the JSON file.
Returns:
Any: Parsed JSON value.
"""
with path.open("r", encoding="utf-8") as stream:
return json.load(stream)
def _as_float(value: Any) -> float | None:
"""Convert a value to a finite float.
Args:
value (Any): Value to convert.
Returns:
float | None: Converted finite float, or ``None`` if conversion fails.
"""
try:
result = float(value)
except (TypeError, ValueError):
return None
if math.isfinite(result):
return result
return None
def _extract_hyperfine(data: dict[str, Any]) -> dict[str, Benchmark]:
"""Extract normalized benchmarks from hyperfine JSON.
Args:
data (dict[str, Any]): Parsed hyperfine JSON object.
Returns:
dict[str, Benchmark]: Benchmarks keyed by command name.
"""
benchmarks: dict[str, Benchmark] = {}
for result in data.get("results", []):
if not isinstance(result, dict):
continue
name = str(result.get("command") or result.get("name") or "").strip()
metric = "median"
value = _as_float(result.get(metric))
if value is None:
metric = "mean"
value = _as_float(result.get(metric))
if name and value is not None:
benchmarks[name] = Benchmark(name=name, value=value, unit="s", metric=metric)
return benchmarks
def _extract_pytest_benchmark(data: dict[str, Any]) -> dict[str, Benchmark]:
"""Extract normalized benchmarks from pytest-benchmark JSON.
Args:
data (dict[str, Any]): Parsed pytest-benchmark JSON object.
Returns:
dict[str, Benchmark]: Benchmarks keyed by full benchmark name.
"""
benchmarks: dict[str, Benchmark] = {}
for benchmark in data.get("benchmarks", []):
if not isinstance(benchmark, dict):
continue
name = str(benchmark.get("fullname") or benchmark.get("name") or "").strip()
stats = benchmark.get("stats", {})
value = None
metric = "median"
if isinstance(stats, dict):
value = _as_float(stats.get(metric))
if value is None:
metric = "mean"
value = _as_float(stats.get(metric))
if name and value is not None:
benchmarks[name] = Benchmark(name=name, value=value, unit="s", metric=metric)
return benchmarks
def _extract_simple_mapping(data: dict[str, Any]) -> dict[str, Benchmark]:
"""Extract normalized benchmarks from a compact mapping JSON object.
Args:
data (dict[str, Any]): Parsed mapping where each benchmark is either a
raw number or an object containing ``value``, ``unit``, and ``metric``.
Returns:
dict[str, Benchmark]: Benchmarks keyed by mapping key.
"""
benchmarks: dict[str, Benchmark] = {}
for name, raw_value in data.items():
if name in {"version", "context", "commit", "timestamp"}:
continue
value = _as_float(raw_value)
unit = ""
metric = "value"
if value is None and isinstance(raw_value, dict):
value = _as_float(raw_value.get("value"))
unit = str(raw_value.get("unit") or "")
metric = str(raw_value.get("metric") or "value")
if value is not None:
benchmarks[str(name)] = Benchmark(name=str(name), value=value, unit=unit, metric=metric)
return benchmarks
def extract_benchmarks(path: Path) -> dict[str, Benchmark]:
"""Extract normalized benchmarks from a supported JSON file.
Args:
path (Path): Path to a hyperfine, pytest-benchmark, or compact mapping
JSON file.
Returns:
dict[str, Benchmark]: Normalized benchmarks keyed by name.
Raises:
ValueError: If the JSON root is not an object or no supported benchmark
entries can be extracted.
"""
data = _read_json(path)
if not isinstance(data, dict):
raise ValueError(f"{path} must contain a JSON object")
extractors = (_extract_hyperfine, _extract_pytest_benchmark, _extract_simple_mapping)
for extractor in extractors:
benchmarks = extractor(data)
if benchmarks:
return benchmarks
raise ValueError(f"No supported benchmark entries found in {path}")
def compare_benchmarks(
baseline: dict[str, Benchmark],
current: dict[str, Benchmark],
threshold_percent: float,
higher_is_better: bool,
) -> tuple[list[Comparison], list[str], list[str]]:
"""Compare baseline benchmarks with current benchmarks.
Args:
baseline (dict[str, Benchmark]): Baseline benchmarks keyed by name.
current (dict[str, Benchmark]): Current benchmarks keyed by name.
threshold_percent (float): Regression threshold in percent.
higher_is_better (bool): If ``True``, lower current values are treated as
regressions. If ``False``, higher current values are treated as
regressions.
Returns:
tuple[list[Comparison], list[str], list[str]]: Comparisons for common
benchmark names, names missing from current results, and names newly
present in current results.
"""
comparisons: list[Comparison] = []
missing_in_current: list[str] = []
new_in_current: list[str] = []
for name, baseline_benchmark in sorted(baseline.items()):
current_benchmark = current.get(name)
if current_benchmark is None:
missing_in_current.append(name)
continue
if baseline_benchmark.value == 0:
delta_percent = 0.0
else:
delta_percent = (
(current_benchmark.value - baseline_benchmark.value)
/ abs(baseline_benchmark.value)
* 100
)
if higher_is_better:
regressed = delta_percent <= -threshold_percent
improved = delta_percent >= threshold_percent
else:
regressed = delta_percent >= threshold_percent
improved = delta_percent <= -threshold_percent
comparisons.append(
Comparison(
name=name,
baseline=baseline_benchmark.value,
current=current_benchmark.value,
delta_percent=delta_percent,
unit=current_benchmark.unit or baseline_benchmark.unit,
metric=current_benchmark.metric,
regressed=regressed,
improved=improved,
)
)
for name in sorted(set(current) - set(baseline)):
new_in_current.append(name)
return comparisons, missing_in_current, new_in_current
def _format_value(value: float, unit: str) -> str:
"""Format a benchmark value for Markdown output.
Args:
value (float): Numeric benchmark value.
unit (str): Display unit.
Returns:
str: Formatted value with optional unit suffix.
"""
suffix = f" {unit}" if unit else ""
return f"{value:.6g}{suffix}"
def _format_status(comparison: Comparison) -> str:
"""Format a comparison status for Markdown output."""
if comparison.regressed:
return ":red_circle: regressed"
if comparison.improved:
return ":green_circle: improved"
return "ok"
def write_summary(
path: Path,
comparisons: list[Comparison],
missing_in_current: list[str],
new_in_current: list[str],
threshold_percent: float,
higher_is_better: bool,
) -> None:
"""Write a Markdown benchmark comparison summary.
Args:
path (Path): Path where the summary should be written.
comparisons (list[Comparison]): Comparison rows for matching benchmarks.
missing_in_current (list[str]): Baseline benchmark names missing from the
current result.
new_in_current (list[str]): Current benchmark names not present in the
baseline result.
threshold_percent (float): Regression threshold in percent.
higher_is_better (bool): Whether higher benchmark values are considered
better.
"""
regressions = [comparison for comparison in comparisons if comparison.regressed]
improvements = [comparison for comparison in comparisons if comparison.improved]
direction = "higher is better" if higher_is_better else "lower is better"
sorted_comparisons = sorted(comparisons, key=lambda comparison: comparison.name)
lines = [
"<!-- bw-benchmark-comment -->",
"## Benchmark comparison",
"",
f"Threshold: {threshold_percent:g}% ({direction}).",
f"Result: {len(regressions)} regression(s), {len(improvements)} improvement(s) beyond threshold.",
]
lines.append("")
if regressions:
lines.extend(
[
f"{len(regressions)} benchmark(s) regressed beyond the configured threshold.",
"",
"| Benchmark | Baseline | Current | Change |",
"| --- | ---: | ---: | ---: |",
]
)
for comparison in regressions:
lines.append(
"| "
f"{comparison.name} | "
f"{_format_value(comparison.baseline, comparison.unit)} | "
f"{_format_value(comparison.current, comparison.unit)} | "
f"{comparison.delta_percent:+.2f}% |"
)
else:
lines.append("No benchmark regression exceeded the configured threshold.")
lines.append("")
if improvements:
lines.extend(
[
f"{len(improvements)} benchmark(s) improved beyond the configured threshold.",
"",
"| Benchmark | Baseline | Current | Change |",
"| --- | ---: | ---: | ---: |",
]
)
for comparison in improvements:
lines.append(
"| "
f"{comparison.name} | "
f"{_format_value(comparison.baseline, comparison.unit)} | "
f"{_format_value(comparison.current, comparison.unit)} | "
f"{comparison.delta_percent:+.2f}% |"
)
else:
lines.append("No benchmark improvement exceeded the configured threshold.")
if sorted_comparisons:
lines.extend(
[
"",
"<details>",
"<summary>All benchmark results</summary>",
"",
"| Benchmark | Baseline | Current | Change | Status |",
"| --- | ---: | ---: | ---: | --- |",
]
)
for comparison in sorted_comparisons:
lines.append(
"| "
f"{comparison.name} | "
f"{_format_value(comparison.baseline, comparison.unit)} | "
f"{_format_value(comparison.current, comparison.unit)} | "
f"{comparison.delta_percent:+.2f}% | "
f"{_format_status(comparison)} |"
)
lines.extend(["", "</details>"])
if missing_in_current:
lines.extend(["", "Missing benchmarks in the current run:"])
lines.extend(f"- `{name}`" for name in missing_in_current)
if new_in_current:
lines.extend(["", "New benchmarks in the current run:"])
lines.extend(f"- `{name}`" for name in new_in_current)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def main() -> int:
"""Run the benchmark comparison command line interface.
Returns:
int: ``1`` when a regression exceeds the threshold, otherwise ``0``.
"""
parser = argparse.ArgumentParser()
parser.add_argument("--baseline", required=True, type=Path)
parser.add_argument("--current", required=True, type=Path)
parser.add_argument("--summary", required=True, type=Path)
parser.add_argument("--threshold-percent", required=True, type=float)
parser.add_argument("--higher-is-better", action="store_true")
args = parser.parse_args()
baseline = extract_benchmarks(args.baseline)
current = extract_benchmarks(args.current)
comparisons, missing_in_current, new_in_current = compare_benchmarks(
baseline=baseline,
current=current,
threshold_percent=args.threshold_percent,
higher_is_better=args.higher_is_better,
)
write_summary(
path=args.summary,
comparisons=comparisons,
missing_in_current=missing_in_current,
new_in_current=new_in_current,
threshold_percent=args.threshold_percent,
higher_is_better=args.higher_is_better,
)
return 1 if any(comparison.regressed for comparison in comparisons) else 0
if __name__ == "__main__":
raise SystemExit(main())
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env bash
##########################
### AI-generated file. ###
##########################
set -euo pipefail
mkdir -p benchmark-results
benchmark_json="${BENCHMARK_JSON:-benchmark-results/current.json}"
benchmark_root="$(dirname "$benchmark_json")"
hyperfine_benchmark_dir="${BENCHMARK_HYPERFINE_DIR:-tests/benchmarks/hyperfine}"
pytest_benchmark_dirs="${BENCHMARK_PYTEST_DIRS:-${BENCHMARK_PYTEST_DIR:-}}"
benchmark_work_dir="$benchmark_root/raw-results"
hyperfine_json_dir="$benchmark_work_dir/hyperfine"
pytest_json="$benchmark_work_dir/pytest.json"
shopt -s nullglob
benchmark_scripts=()
benchmark_scripts=("$hyperfine_benchmark_dir"/benchmark_*.sh)
shopt -u nullglob
pytest_dirs=()
for pytest_benchmark_dir in $pytest_benchmark_dirs; do
if [ -d "$pytest_benchmark_dir" ]; then
pytest_dirs+=("$pytest_benchmark_dir")
else
echo "Pytest benchmark directory not found: $pytest_benchmark_dir" >&2
exit 1
fi
done
if [ "${#benchmark_scripts[@]}" -eq 0 ] && [ "${#pytest_dirs[@]}" -eq 0 ]; then
echo "No benchmark scripts or pytest benchmarks found" >&2
exit 1
fi
echo "Benchmark Python: $(command -v python)"
python -c 'import sys; print(sys.version)'
rm -rf "$benchmark_work_dir"
mkdir -p "$hyperfine_json_dir"
if [ "${#benchmark_scripts[@]}" -gt 0 ]; then
for benchmark_script in "${benchmark_scripts[@]}"; do
title="$(sed -n 's/^# BENCHMARK_TITLE:[[:space:]]*//p' "$benchmark_script" | head -n 1)"
if [ -z "$title" ]; then
title="$(basename "$benchmark_script" .sh)"
fi
benchmark_name="$(basename "$benchmark_script" .sh)"
benchmark_result_json="$hyperfine_json_dir/$benchmark_name.json"
echo "Preflight benchmark script: $benchmark_script"
bash "$benchmark_script"
hyperfine \
--show-output \
--warmup 1 \
--runs 5 \
--command-name "$title" \
--export-json "$benchmark_result_json" \
"bash $(printf "%q" "$benchmark_script")"
done
fi
if [ "${#pytest_dirs[@]}" -gt 0 ]; then
pytest \
-q "${pytest_dirs[@]}" \
--benchmark-only \
--benchmark-json "$pytest_json"
fi
python .github/scripts/aggregate_benchmarks.py \
--input-dir "$benchmark_work_dir" \
--output "$benchmark_json"
+125
View File
@@ -0,0 +1,125 @@
##########################
### AI-generated file. ###
##########################
"""Run a command with BEC e2e services available."""
from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import tempfile
import time
from pathlib import Path
import bec_lib
from bec_ipython_client import BECIPythonClient
from bec_lib.redis_connector import RedisConnector
from bec_lib.service_config import ServiceConfig, ServiceConfigModel
from redis import Redis
def _wait_for_redis(host: str, port: int) -> None:
client = Redis(host=host, port=port)
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
try:
if client.ping():
return
except Exception:
time.sleep(0.1)
raise RuntimeError(f"Redis did not start on {host}:{port}")
def _start_redis(files_path: Path, host: str, port: int) -> subprocess.Popen:
redis_server = shutil.which("redis-server")
if redis_server is None:
raise RuntimeError("redis-server executable not found")
return subprocess.Popen(
[
redis_server,
"--bind",
host,
"--port",
str(port),
"--save",
"",
"--appendonly",
"no",
"--dir",
str(files_path),
]
)
def _write_configs(files_path: Path, host: str, port: int) -> Path:
test_config = files_path / "test_config.yaml"
services_config = files_path / "services_config.yaml"
bec_lib_path = Path(bec_lib.__file__).resolve().parent
shutil.copyfile(bec_lib_path / "tests" / "test_config.yaml", test_config)
service_config = ServiceConfigModel(
redis={"host": host, "port": port}, file_writer={"base_path": str(files_path)}
)
services_config.write_text(service_config.model_dump_json(indent=4), encoding="utf-8")
return services_config
def _load_demo_config(services_config: Path) -> None:
bec = BECIPythonClient(ServiceConfig(services_config), RedisConnector, forced=True)
bec.start()
try:
bec.config.load_demo_config()
finally:
bec.shutdown()
bec._client._reset_singleton()
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args()
if args.command[:1] == ["--"]:
args.command = args.command[1:]
if not args.command:
raise ValueError("No command provided")
host = "127.0.0.1"
port = 6379
with tempfile.TemporaryDirectory(prefix="bec-benchmark-") as tmp:
files_path = Path(tmp)
services_config = _write_configs(files_path, host, port)
redis_process = _start_redis(files_path, host, port)
processes = None
service_handler = None
try:
_wait_for_redis(host, port)
from bec_server.bec_server_utils.service_handler import ServiceHandler
service_handler = ServiceHandler(
bec_path=files_path, config_path=services_config, interface="subprocess"
)
processes = service_handler.start()
_load_demo_config(services_config)
env = os.environ.copy()
return subprocess.run(args.command, env=env, check=False).returncode
finally:
if service_handler is not None and processes is not None:
service_handler.stop(processes)
redis_process.terminate()
try:
redis_process.wait(timeout=10)
except subprocess.TimeoutExpired:
redis_process.kill()
if __name__ == "__main__":
raise SystemExit(main())
+242
View File
@@ -0,0 +1,242 @@
name: BW Benchmarks
on: [ workflow_call ]
permissions:
contents: read
env:
BENCHMARK_JSON: benchmark-results/current.json
BENCHMARK_BASELINE_JSON: gh-pages-benchmark-data/benchmarks/latest.json
BENCHMARK_SUMMARY: benchmark-results/summary.md
BENCHMARK_COMMAND: "bash .github/scripts/run_benchmarks.sh"
BENCHMARK_THRESHOLD_PERCENT: 20
BENCHMARK_HIGHER_IS_BETTER: false
jobs:
benchmark_attempt:
runs-on: ubuntu-latest
continue-on-error: true
permissions:
contents: read
defaults:
run:
shell: bash -el {0}
strategy:
fail-fast: false
matrix:
attempt: [ 1, 2, 3 ]
env:
BENCHMARK_JSON: benchmark-results/current-${{ matrix.attempt }}.json
BEC_CORE_BRANCH: main
OPHYD_DEVICES_BRANCH: main
PLUGIN_REPO_BRANCH: main
BENCHMARK_PYTEST_DIRS: tests/unit_tests/benchmarks
QTWEBENGINE_DISABLE_SANDBOX: 1
QT_QPA_PLATFORM: "offscreen"
steps:
- name: Checkout BEC Widgets
uses: actions/checkout@v4
with:
repository: bec-project/bec_widgets
ref: ${{ github.event.pull_request.head.sha || github.sha }}
- name: Set up Conda
uses: conda-incubator/setup-miniconda@v3
with:
auto-update-conda: true
auto-activate-base: true
python-version: "3.11"
- name: Install system dependencies
run: |
sudo apt-get update
sudo apt-get install -y libgl1 libegl1 x11-utils libxkbcommon-x11-0 libdbus-1-3 xvfb
sudo apt-get -y install libnss3 libxdamage1 libasound2t64 libatomic1 libxcursor1
sudo apt-get -y install ttyd hyperfine redis-server
- name: Install full e2e environment
run: |
echo -e "\033[35;1m Using branch $BEC_CORE_BRANCH of BEC CORE \033[0;m";
git clone --branch "$BEC_CORE_BRANCH" https://github.com/bec-project/bec.git
echo -e "\033[35;1m Using branch $OPHYD_DEVICES_BRANCH of OPHYD_DEVICES \033[0;m";
git clone --branch "$OPHYD_DEVICES_BRANCH" https://github.com/bec-project/ophyd_devices.git
export OHPYD_DEVICES_PATH=$PWD/ophyd_devices
echo -e "\033[35;1m Using branch $PLUGIN_REPO_BRANCH of bec_testing_plugin \033[0;m";
git clone --branch "$PLUGIN_REPO_BRANCH" https://github.com/bec-project/bec_testing_plugin.git
cd ./bec
conda create -q -n test-environment python=3.11
conda activate test-environment
source ./bin/install_bec_dev.sh -t
cd ../
python -m pip install -e ./ophyd_devices -e .[dev,pyside6] -e ./bec_testing_plugin pytest-benchmark
mkdir -p "$(dirname "$BENCHMARK_JSON")"
python .github/scripts/run_with_bec_servers.py -- bash -lc "$BENCHMARK_COMMAND"
test -s "$BENCHMARK_JSON"
- name: Upload benchmark artifact
uses: actions/upload-artifact@v4
with:
name: bw-benchmark-json-${{ matrix.attempt }}
path: ${{ env.BENCHMARK_JSON }}
benchmark:
needs: [ benchmark_attempt ]
runs-on: ubuntu-latest
permissions:
contents: read
issues: write
pull-requests: write
steps:
- name: Checkout BEC Widgets
uses: actions/checkout@v4
with:
repository: bec-project/bec_widgets
ref: ${{ github.event.pull_request.head.sha || github.sha }}
- name: Download benchmark attempts
uses: actions/download-artifact@v4
with:
pattern: bw-benchmark-json-*
path: benchmark-results/attempts
merge-multiple: true
- name: Aggregate benchmark attempts
run: |
python .github/scripts/aggregate_benchmarks.py \
--input-dir benchmark-results/attempts \
--output "$BENCHMARK_JSON"
- name: Upload aggregate benchmark artifact
uses: actions/upload-artifact@v4
with:
name: bw-benchmark-json
path: ${{ env.BENCHMARK_JSON }}
- name: Fetch gh-pages benchmark data
run: |
if git ls-remote --exit-code --heads origin gh-pages; then
git clone --depth=1 --branch gh-pages "$GITHUB_SERVER_URL/$GITHUB_REPOSITORY.git" gh-pages-benchmark-data
else
mkdir -p gh-pages-benchmark-data
fi
- name: Compare with latest gh-pages benchmark
id: compare
continue-on-error: true
run: |
if [ ! -s "$BENCHMARK_BASELINE_JSON" ]; then
mkdir -p "$(dirname "$BENCHMARK_SUMMARY")"
{
echo "<!-- bw-benchmark-comment -->"
echo "## Benchmark comparison"
echo
echo "No benchmark baseline was found on gh-pages."
} > "$BENCHMARK_SUMMARY"
exit 0
fi
args=(
--baseline "$BENCHMARK_BASELINE_JSON"
--current "$BENCHMARK_JSON"
--summary "$BENCHMARK_SUMMARY"
--threshold-percent "$BENCHMARK_THRESHOLD_PERCENT"
)
if [ "$BENCHMARK_HIGHER_IS_BETTER" = "true" ]; then
args+=(--higher-is-better)
fi
set +e
python .github/scripts/compare_benchmarks.py "${args[@]}"
status=$?
set -e
if [ ! -s "$BENCHMARK_SUMMARY" ]; then
mkdir -p "$(dirname "$BENCHMARK_SUMMARY")"
{
echo "<!-- bw-benchmark-comment -->"
echo "## Benchmark comparison"
echo
echo "Benchmark comparison failed before writing a summary."
} > "$BENCHMARK_SUMMARY"
fi
exit "$status"
- name: Find existing benchmark PR comment
if: github.event_name == 'pull_request'
id: fc
uses: peter-evans/find-comment@v3
with:
issue-number: ${{ github.event.pull_request.number }}
comment-author: github-actions[bot]
body-includes: "<!-- bw-benchmark-comment -->"
- name: Create or update benchmark PR comment
if: github.event_name == 'pull_request'
uses: peter-evans/create-or-update-comment@v5
with:
issue-number: ${{ github.event.pull_request.number }}
comment-id: ${{ steps.fc.outputs.comment-id }}
body-path: ${{ env.BENCHMARK_SUMMARY }}
edit-mode: replace
- name: Fail on benchmark regression
if: github.event_name == 'pull_request' && steps.compare.outcome == 'failure'
run: exit 1
publish:
needs: [ benchmark ]
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
permissions:
contents: write
steps:
- name: Checkout BEC Widgets
uses: actions/checkout@v4
with:
repository: bec-project/bec_widgets
ref: ${{ github.sha }}
- name: Download aggregate benchmark artifact
uses: actions/download-artifact@v4
with:
name: bw-benchmark-json
path: benchmark-results
- name: Verify aggregate benchmark artifact
run: test -s "$BENCHMARK_JSON"
- name: Prepare gh-pages for publishing
run: |
# Clean up any existing worktree/directory
if [ -d gh-pages-benchmark-data ]; then
git worktree remove gh-pages-benchmark-data --force || rm -rf gh-pages-benchmark-data
fi
if git ls-remote --exit-code --heads origin gh-pages; then
git fetch --depth=1 origin gh-pages
git worktree add gh-pages-benchmark-data FETCH_HEAD
else
git worktree add --detach gh-pages-benchmark-data
git -C gh-pages-benchmark-data checkout --orphan gh-pages
git -C gh-pages-benchmark-data rm -rf .
fi
- name: Publish benchmark data to gh-pages
working-directory: gh-pages-benchmark-data
run: |
mkdir -p benchmarks/history
cp "../$BENCHMARK_JSON" benchmarks/latest.json
cp "../$BENCHMARK_JSON" "benchmarks/history/${GITHUB_SHA}.json"
git config user.name "github-actions[bot]"
git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
git add benchmarks/latest.json "benchmarks/history/${GITHUB_SHA}.json"
git commit -m "Update BW benchmark data for ${GITHUB_SHA}" || exit 0
git push origin HEAD:gh-pages
+17 -7
View File
@@ -1,19 +1,19 @@
name: Full CI
on:
on:
push:
pull_request:
workflow_dispatch:
inputs:
BEC_WIDGETS_BRANCH:
description: 'Branch of BEC Widgets to install'
description: "Branch of BEC Widgets to install"
required: false
type: string
BEC_CORE_BRANCH:
description: 'Branch of BEC Core to install'
description: "Branch of BEC Core to install"
required: false
type: string
OPHYD_DEVICES_BRANCH:
description: 'Branch of Ophyd Devices to install'
description: "Branch of Ophyd Devices to install"
required: false
type: string
@@ -23,6 +23,7 @@ concurrency:
permissions:
pull-requests: write
contents: read
jobs:
check_pr_status:
@@ -33,6 +34,15 @@ jobs:
if: needs.check_pr_status.outputs.branch-pr == ''
uses: ./.github/workflows/formatter.yml
benchmark:
needs: [check_pr_status]
if: needs.check_pr_status.outputs.branch-pr == ''
permissions:
contents: write
issues: write
pull-requests: write
uses: ./.github/workflows/benchmark.yml
unit-test:
needs: [check_pr_status, formatter]
if: needs.check_pr_status.outputs.branch-pr == ''
@@ -69,9 +79,9 @@ jobs:
uses: ./.github/workflows/child_repos.yml
with:
BEC_CORE_BRANCH: ${{ inputs.BEC_CORE_BRANCH || 'main' }}
OPHYD_DEVICES_BRANCH: ${{ inputs.OPHYD_DEVICES_BRANCH || 'main'}}
OPHYD_DEVICES_BRANCH: ${{ inputs.OPHYD_DEVICES_BRANCH || 'main'}}
BEC_WIDGETS_BRANCH: ${{ inputs.BEC_WIDGETS_BRANCH || github.head_ref || github.sha }}
plugin_repos:
needs: [check_pr_status, formatter]
if: needs.check_pr_status.outputs.branch-pr == ''
@@ -81,4 +91,4 @@ jobs:
BEC_WIDGETS_BRANCH: ${{ inputs.BEC_WIDGETS_BRANCH || github.head_ref || github.sha }}
secrets:
GH_READ_TOKEN: ${{ secrets.GH_READ_TOKEN }}
GH_READ_TOKEN: ${{ secrets.GH_READ_TOKEN }}
+12 -13
View File
@@ -1,25 +1,25 @@
name: Run Pytest with different Python versions
on:
on:
workflow_call:
inputs:
pr_number:
description: 'Pull request number'
description: "Pull request number"
required: false
type: number
BEC_CORE_BRANCH:
description: 'Branch of BEC Core to install'
description: "Branch of BEC Core to install"
required: false
default: 'main'
default: "main"
type: string
OPHYD_DEVICES_BRANCH:
description: 'Branch of Ophyd Devices to install'
description: "Branch of Ophyd Devices to install"
required: false
default: 'main'
default: "main"
type: string
BEC_WIDGETS_BRANCH:
description: 'Branch of BEC Widgets to install'
description: "Branch of BEC Widgets to install"
required: false
default: 'main'
default: "main"
type: string
jobs:
@@ -30,15 +30,14 @@ jobs:
python-version: ["3.11", "3.12", "3.13"]
env:
BEC_WIDGETS_BRANCH: main # Set the branch you want for bec_widgets
BEC_CORE_BRANCH: main # Set the branch you want for bec
OPHYD_DEVICES_BRANCH: main # Set the branch you want for ophyd_devices
BEC_WIDGETS_BRANCH: main # Set the branch you want for bec_widgets
BEC_CORE_BRANCH: main # Set the branch you want for bec
OPHYD_DEVICES_BRANCH: main # Set the branch you want for ophyd_devices
PROJECT_PATH: ${{ github.repository }}
QTWEBENGINE_DISABLE_SANDBOX: 1
QT_QPA_PLATFORM: "offscreen"
steps:
- name: Checkout BEC Widgets
uses: actions/checkout@v4
with:
@@ -56,4 +55,4 @@ jobs:
- name: Run Pytest
run: |
pip install pytest pytest-random-order
pytest -v --junitxml=report.xml --random-order ./tests/unit_tests
pytest -v --junitxml=report.xml --random-order --ignore=tests/unit_tests/benchmarks ./tests/unit_tests
+10 -12
View File
@@ -1,32 +1,30 @@
name: Run Pytest with Coverage
on:
on:
workflow_call:
inputs:
pr_number:
description: 'Pull request number'
description: "Pull request number"
required: false
type: number
BEC_CORE_BRANCH:
description: 'Branch of BEC Core to install'
description: "Branch of BEC Core to install"
required: false
default: 'main'
default: "main"
type: string
OPHYD_DEVICES_BRANCH:
description: 'Branch of Ophyd Devices to install'
description: "Branch of Ophyd Devices to install"
required: false
default: 'main'
default: "main"
type: string
BEC_WIDGETS_BRANCH:
description: 'Branch of BEC Widgets to install'
description: "Branch of BEC Widgets to install"
required: false
default: 'main'
default: "main"
type: string
secrets:
CODECOV_TOKEN:
required: true
permissions:
pull-requests: write
@@ -55,7 +53,7 @@ jobs:
- name: Run Pytest with Coverage
id: coverage
run: pytest --random-order --cov=bec_widgets --cov-config=pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail tests/unit_tests/
run: pytest --random-order --cov=bec_widgets --cov-config=pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail --ignore=tests/unit_tests/benchmarks tests/unit_tests/
- name: Upload test artifacts
uses: actions/upload-artifact@v4
@@ -69,4 +67,4 @@ jobs:
uses: codecov/codecov-action@v5
with:
token: ${{ secrets.CODECOV_TOKEN }}
slug: bec-project/bec_widgets
slug: bec-project/bec_widgets
+89
View File
@@ -1,6 +1,95 @@
# CHANGELOG
## v3.8.0 (2026-05-01)
### Bug Fixes
- **dock_area**: Change to show_dialo=False for CLI profile baseline restore
([`0b1f0b4`](https://github.com/bec-project/bec_widgets/commit/0b1f0b4c262ff31469b7114b9f00bf0a7b85e8f2))
- **dock_area**: Cli call load_profile has restore_baseline kwarg
([`cc82597`](https://github.com/bec-project/bec_widgets/commit/cc825972c202cd9ded32f8b2d1ce5f822c2ebdba))
### Features
- **dock_area**: Add CLI restore current profile from baseline with optional confirmation dialog
([`17865a2`](https://github.com/bec-project/bec_widgets/commit/17865a2c338a4a1f944659dde4ec05c25a8dd963))
## v3.7.3 (2026-05-01)
### Bug Fixes
- **dock_area**: Profile names changed, default->baseline, user->runtime
([`dd32caf`](https://github.com/bec-project/bec_widgets/commit/dd32caf6e815fd1922b6aae84d00decad9dbf869))
### Testing
- **dock_area**: Remove low-value tests
([`717d74b`](https://github.com/bec-project/bec_widgets/commit/717d74b19e8c6960209190c47ba32732ffaa0094))
## v3.7.2 (2026-04-29)
### Bug Fixes
- **dock-area**: Avoid switching profile when saving new profile
([`73b44cf`](https://github.com/bec-project/bec_widgets/commit/73b44cffb219347cacb609f3b93068eda6701b42))
- **workspace-actions**: Use try/finally and restore previous blocked state in refresh_profiles
([`30ef255`](https://github.com/bec-project/bec_widgets/commit/30ef25533af9df5a9bd9e69808dc49fdf22f4318))
Agent-Logs-Url:
https://github.com/bec-project/bec_widgets/sessions/004cb4bc-5015-485e-a803-1e63876b7024
Co-authored-by: wyzula-jan <133381102+wyzula-jan@users.noreply.github.com>
### Build System
- Add pytest-benchmark dependency
([`551d38d`](https://github.com/bec-project/bec_widgets/commit/551d38d90111361e64bfcf10a1409be71dd298bc))
### Chores
- Update header comments in script files to indicate AI generation
([`3f1aa80`](https://github.com/bec-project/bec_widgets/commit/3f1aa80756368454c3631cb6cf9d29db28af177a))
### Continuous Integration
- Add benchmark workflow
([`999b7a2`](https://github.com/bec-project/bec_widgets/commit/999b7a2321f2f222c04b056a2db4280f66de9c48))
- Fix benchmark upload
([`19b5c8f`](https://github.com/bec-project/bec_widgets/commit/19b5c8f724dbdb7421957c290f9213bf072392df))
- Increase threshold to 20 percent
([`409c9e5`](https://github.com/bec-project/bec_widgets/commit/409c9e5bfacdfc003f1bc8c9944f01798bd3818e))
### Testing
- Fix assertions after updating ophyd devices templates
([`a614d66`](https://github.com/bec-project/bec_widgets/commit/a614d662d6da716a49afb4ed3a3903f108210386))
Co-authored-by: Copilot <copilot@github.com>
- Remove references to "scan_motors" in tests
([`5056ef8`](https://github.com/bec-project/bec_widgets/commit/5056ef8946d03a20e802709d3fe81c84c195fe41))
## v3.7.1 (2026-04-21)
### Bug Fixes
- **heatmap**: Fix access to status from metadata
([`55694ff`](https://github.com/bec-project/bec_widgets/commit/55694ff2b96581e03c63c8b8e068e2db79bcf780))
### Testing
- Fix exit status and status access in tests
([`91afc77`](https://github.com/bec-project/bec_widgets/commit/91afc775d59b4ba31bed3585847a67c301acf9b0))
## v3.7.0 (2026-04-21)
### Features
+46 -14
View File
@@ -340,10 +340,10 @@ class BECDockArea(RPCBase):
Save the current workspace profile.
On first save of a given name:
- writes a default copy to states/default/<name>.ini with tag=default and created_at
- writes a user copy to states/user/<name>.ini with tag=user and created_at
On subsequent saves of user-owned profiles:
- updates both the default and user copies so restore uses the latest snapshot.
- writes a baseline copy to profiles/baseline/<name>.ini with created_at
- writes a runtime copy to profiles/runtime/<name>.ini with created_at
On subsequent saves:
- updates both the baseline and runtime copies so restore uses the latest snapshot.
Read-only bundled profiles cannot be overwritten.
Args:
@@ -358,15 +358,31 @@ class BECDockArea(RPCBase):
@rpc_timeout(None)
@rpc_call
def load_profile(self, name: "str | None" = None):
def load_profile(self, name: "str | None" = None, restore_baseline: "bool" = False):
"""
Load a workspace profile.
Before switching, persist the current profile to the user copy.
Prefer loading the user copy; fall back to the default copy.
Before switching, persist the current profile to the runtime copy.
Prefer loading the runtime copy; fall back to the baseline copy. When
``restore_baseline`` is True, first overwrite the runtime copy with the
baseline profile and then load it.
Args:
name (str | None): The name of the profile to load. If None, prompts the user.
restore_baseline (bool): If True, restore the runtime copy from the
baseline before loading. Defaults to False.
"""
@rpc_timeout(None)
@rpc_call
def restore_baseline_profile(self, name: "str | None" = None, show_dialog: "bool" = False):
"""
Overwrite the runtime copy of *name* with the baseline.
If *name* is None, target the currently active profile.
Args:
name (str | None): The name of the profile to restore. If None, uses the current profile.
show_dialog (bool): If True, ask for confirmation before restoring.
"""
@rpc_call
@@ -1348,10 +1364,10 @@ class DockAreaView(RPCBase):
Save the current workspace profile.
On first save of a given name:
- writes a default copy to states/default/<name>.ini with tag=default and created_at
- writes a user copy to states/user/<name>.ini with tag=user and created_at
On subsequent saves of user-owned profiles:
- updates both the default and user copies so restore uses the latest snapshot.
- writes a baseline copy to profiles/baseline/<name>.ini with created_at
- writes a runtime copy to profiles/runtime/<name>.ini with created_at
On subsequent saves:
- updates both the baseline and runtime copies so restore uses the latest snapshot.
Read-only bundled profiles cannot be overwritten.
Args:
@@ -1366,15 +1382,31 @@ class DockAreaView(RPCBase):
@rpc_timeout(None)
@rpc_call
def load_profile(self, name: "str | None" = None):
def load_profile(self, name: "str | None" = None, restore_baseline: "bool" = False):
"""
Load a workspace profile.
Before switching, persist the current profile to the user copy.
Prefer loading the user copy; fall back to the default copy.
Before switching, persist the current profile to the runtime copy.
Prefer loading the runtime copy; fall back to the baseline copy. When
``restore_baseline`` is True, first overwrite the runtime copy with the
baseline profile and then load it.
Args:
name (str | None): The name of the profile to load. If None, prompts the user.
restore_baseline (bool): If True, restore the runtime copy from the
baseline before loading. Defaults to False.
"""
@rpc_timeout(None)
@rpc_call
def restore_baseline_profile(self, name: "str | None" = None, show_dialog: "bool" = False):
"""
Overwrite the runtime copy of *name* with the baseline.
If *name* is None, target the currently active profile.
Args:
name (str | None): The name of the profile to restore. If None, uses the current profile.
show_dialog (bool): If True, ask for confirmation before restoring.
"""
@rpc_call
@@ -35,25 +35,25 @@ from bec_widgets.utils.widget_state_manager import WidgetStateManager
from bec_widgets.widgets.containers.dock_area.basic_dock_area import DockAreaWidget
from bec_widgets.widgets.containers.dock_area.profile_utils import (
SETTINGS_KEYS,
default_profile_candidates,
baseline_profile_candidates,
delete_profile_files,
get_last_profile,
is_profile_read_only,
is_quick_select,
list_profiles,
list_quick_profiles,
load_default_profile_screenshot,
load_user_profile_screenshot,
load_baseline_profile_screenshot,
load_runtime_profile_screenshot,
now_iso_utc,
open_default_settings,
open_user_settings,
open_baseline_settings,
open_runtime_settings,
profile_origin,
profile_origin_display,
read_manifest,
restore_user_from_default,
restore_runtime_from_baseline,
runtime_profile_candidates,
set_last_profile,
set_quick_select,
user_profile_candidates,
write_manifest,
)
from bec_widgets.widgets.containers.dock_area.settings.dialogs import (
@@ -108,6 +108,7 @@ class BECDockArea(DockAreaWidget):
"list_profiles",
"save_profile",
"load_profile",
"restore_baseline_profile",
"delete_profile",
]
@@ -235,11 +236,8 @@ class BECDockArea(DockAreaWidget):
def _load_initial_profile(self, name: str) -> None:
"""Load the initial profile."""
self.load_profile(name)
combo = self.toolbar.components.get_action("workspace_combo").widget
combo.blockSignals(True)
if not self._empty_profile_active:
combo.setCurrentText(name)
combo.blockSignals(False)
self._set_workspace_combo_text_silent(name)
def _start_empty_workspace(self) -> None:
"""
@@ -591,13 +589,13 @@ class BECDockArea(DockAreaWidget):
@property
def profile_namespace(self) -> str | None:
"""Namespace used to scope user/default profile files for this dock area."""
"""Namespace used to scope runtime/baseline profile files for this dock area."""
return self._resolve_profile_namespace()
def _profile_exists(self, name: str, namespace: str | None) -> bool:
return any(
os.path.exists(path) for path in user_profile_candidates(name, namespace)
) or any(os.path.exists(path) for path in default_profile_candidates(name, namespace))
os.path.exists(path) for path in runtime_profile_candidates(name, namespace)
) or any(os.path.exists(path) for path in baseline_profile_candidates(name, namespace))
def _write_snapshot_to_settings(self, settings, save_preview: bool = True) -> None:
"""
@@ -623,35 +621,34 @@ class BECDockArea(DockAreaWidget):
name: str,
namespace: str | None,
*,
write_default: bool = True,
write_user: bool = True,
write_baseline: bool = True,
write_runtime: bool = True,
save_preview: bool = True,
) -> None:
"""
Write profile settings to default and/or user settings files.
Write profile settings to baseline and/or runtime settings files.
Args:
name: The profile name.
namespace: The profile namespace.
write_default: Whether to write to the default settings file.
write_user: Whether to write to the user settings file.
write_baseline: Whether to write to the baseline settings file.
write_runtime: Whether to write to the runtime settings file.
save_preview: Whether to save a screenshot preview.
"""
if write_default:
ds = open_default_settings(name, namespace=namespace)
self._write_snapshot_to_settings(ds, save_preview=save_preview)
if not ds.value(SETTINGS_KEYS["created_at"], ""):
ds.setValue(SETTINGS_KEYS["created_at"], now_iso_utc())
if not ds.value(SETTINGS_KEYS["is_quick_select"], None):
ds.setValue(SETTINGS_KEYS["is_quick_select"], True)
if write_user:
us = open_user_settings(name, namespace=namespace)
self._write_snapshot_to_settings(us, save_preview=save_preview)
if not us.value(SETTINGS_KEYS["created_at"], ""):
us.setValue(SETTINGS_KEYS["created_at"], now_iso_utc())
if not us.value(SETTINGS_KEYS["is_quick_select"], None):
us.setValue(SETTINGS_KEYS["is_quick_select"], True)
def _write_settings(open_settings) -> None:
settings = open_settings(name, namespace=namespace)
self._write_snapshot_to_settings(settings, save_preview=save_preview)
if not settings.value(SETTINGS_KEYS["created_at"], ""):
settings.setValue(SETTINGS_KEYS["created_at"], now_iso_utc())
if not settings.value(SETTINGS_KEYS["is_quick_select"], None):
settings.setValue(SETTINGS_KEYS["is_quick_select"], True)
if write_baseline:
_write_settings(open_baseline_settings)
if write_runtime:
_write_settings(open_runtime_settings)
def _finalize_profile_change(self, name: str, namespace: str | None) -> None:
"""
@@ -669,6 +666,14 @@ class BECDockArea(DockAreaWidget):
combo = self.toolbar.components.get_action("workspace_combo").widget
combo.refresh_profiles(active_profile=name)
def _set_workspace_combo_text_silent(self, text: str) -> None:
combo = self.toolbar.components.get_action("workspace_combo").widget
was_blocked = combo.blockSignals(True)
try:
combo.setCurrentText(text)
finally:
combo.blockSignals(was_blocked)
def _enter_empty_profile_state(self) -> None:
"""
Switch to the transient empty workspace state.
@@ -705,10 +710,10 @@ class BECDockArea(DockAreaWidget):
Save the current workspace profile.
On first save of a given name:
- writes a default copy to states/default/<name>.ini with tag=default and created_at
- writes a user copy to states/user/<name>.ini with tag=user and created_at
On subsequent saves of user-owned profiles:
- updates both the default and user copies so restore uses the latest snapshot.
- writes a baseline copy to profiles/baseline/<name>.ini with created_at
- writes a runtime copy to profiles/runtime/<name>.ini with created_at
On subsequent saves:
- updates both the baseline and runtime copies so restore uses the latest snapshot.
Read-only bundled profiles cannot be overwritten.
Args:
@@ -772,7 +777,7 @@ class BECDockArea(DockAreaWidget):
overwrite_existing = origin == "settings"
origin_before_save = profile_origin(name, namespace=namespace)
overwrite_default = overwrite_existing and origin_before_save == "settings"
overwrite_baseline = overwrite_existing and origin_before_save == "settings"
# Display saving placeholder in toolbar
workspace_combo = self.toolbar.components.get_action("workspace_combo").widget
@@ -781,12 +786,12 @@ class BECDockArea(DockAreaWidget):
workspace_combo.setCurrentIndex(0)
workspace_combo.blockSignals(False)
# Write to default and/or user settings
should_write_default = overwrite_default or not any(
os.path.exists(path) for path in default_profile_candidates(name, namespace)
# Write to baseline and/or runtime settings
should_write_baseline = overwrite_baseline or not any(
os.path.exists(path) for path in baseline_profile_candidates(name, namespace)
)
self._write_profile_settings(
name, namespace, write_default=should_write_default, write_user=True
name, namespace, write_baseline=should_write_baseline, write_runtime=True
)
set_quick_select(name, quickselect, namespace=namespace)
@@ -796,7 +801,6 @@ class BECDockArea(DockAreaWidget):
self._pending_autosave_skip = (current_profile, name)
else:
self._pending_autosave_skip = None
workspace_combo.setCurrentText(name)
self._finalize_profile_change(name, namespace)
@SafeSlot()
@@ -816,16 +820,21 @@ class BECDockArea(DockAreaWidget):
@SafeSlot()
@SafeSlot(str)
@SafeSlot(str, bool)
@rpc_timeout(None)
def load_profile(self, name: str | None = None):
def load_profile(self, name: str | None = None, restore_baseline: bool = False):
"""
Load a workspace profile.
Before switching, persist the current profile to the user copy.
Prefer loading the user copy; fall back to the default copy.
Before switching, persist the current profile to the runtime copy.
Prefer loading the runtime copy; fall back to the baseline copy. When
``restore_baseline`` is True, first overwrite the runtime copy with the
baseline profile and then load it.
Args:
name (str | None): The name of the profile to load. If None, prompts the user.
restore_baseline (bool): If True, restore the runtime copy from the
baseline before loading. Defaults to False.
"""
if name == "":
return
@@ -844,14 +853,17 @@ class BECDockArea(DockAreaWidget):
if skip_pair and skip_pair == (prev_name, name):
self._pending_autosave_skip = None
else:
us_prev = open_user_settings(prev_name, namespace=namespace)
us_prev = open_runtime_settings(prev_name, namespace=namespace)
self._write_snapshot_to_settings(us_prev, save_preview=True)
if restore_baseline:
restore_runtime_from_baseline(name, namespace=namespace)
settings = None
if any(os.path.exists(path) for path in user_profile_candidates(name, namespace)):
settings = open_user_settings(name, namespace=namespace)
elif any(os.path.exists(path) for path in default_profile_candidates(name, namespace)):
settings = open_default_settings(name, namespace=namespace)
if any(os.path.exists(path) for path in runtime_profile_candidates(name, namespace)):
settings = open_runtime_settings(name, namespace=namespace)
elif any(os.path.exists(path) for path in baseline_profile_candidates(name, namespace)):
settings = open_baseline_settings(name, namespace=namespace)
if settings is None:
logger.warning(f"Profile '{name}' not found in namespace '{namespace}'. Creating new.")
self.delete_all()
@@ -893,32 +905,36 @@ class BECDockArea(DockAreaWidget):
@SafeSlot()
@SafeSlot(str)
def restore_user_profile_from_default(self, name: str | None = None):
@SafeSlot(str, bool)
@rpc_timeout(None)
def restore_baseline_profile(self, name: str | None = None, show_dialog: bool = False):
"""
Overwrite the user copy of *name* with the default baseline.
Overwrite the runtime copy of *name* with the baseline.
If *name* is None, target the currently active profile.
Args:
name (str | None): The name of the profile to restore. If None, uses the current profile.
show_dialog (bool): If True, ask for confirmation before restoring.
"""
target = name or getattr(self, "_current_profile_name", None)
if not target:
return
namespace = self.profile_namespace
current_pixmap = None
if self.isVisible():
current_pixmap = QPixmap()
ba = bytes(self.screenshot_bytes())
current_pixmap.loadFromData(ba)
if current_pixmap is None or current_pixmap.isNull():
current_pixmap = load_user_profile_screenshot(target, namespace=namespace)
default_pixmap = load_default_profile_screenshot(target, namespace=namespace)
if show_dialog:
current_pixmap = None
if self.isVisible():
current_pixmap = QPixmap()
ba = bytes(self.screenshot_bytes())
current_pixmap.loadFromData(ba)
if current_pixmap is None or current_pixmap.isNull():
current_pixmap = load_runtime_profile_screenshot(target, namespace=namespace)
baseline_pixmap = load_baseline_profile_screenshot(target, namespace=namespace)
if not RestoreProfileDialog.confirm(self, current_pixmap, default_pixmap):
return
if not RestoreProfileDialog.confirm(self, current_pixmap, baseline_pixmap):
return
restore_user_from_default(target, namespace=namespace)
restore_runtime_from_baseline(target, namespace=namespace)
self.delete_all()
self.load_profile(target)
@@ -1053,7 +1069,7 @@ class BECDockArea(DockAreaWidget):
manage_action = self.toolbar.components.get_action("manage_workspaces").action
if self.manage_dialog is None or not self.manage_dialog.isVisible():
self.manage_widget = WorkSpaceManager(
self, target_widget=self, default_profile=self._current_profile_name
self, target_widget=self, active_profile=self._current_profile_name
)
self.manage_dialog = QDialog(modal=False)
@@ -1152,7 +1168,7 @@ class BECDockArea(DockAreaWidget):
return
namespace = self.profile_namespace
settings = open_user_settings(name, namespace=namespace)
settings = open_runtime_settings(name, namespace=namespace)
self._write_snapshot_to_settings(settings)
set_last_profile(name, namespace=namespace, instance=self._last_profile_instance_id())
self._exit_snapshot_written = True
@@ -2,9 +2,13 @@
Utilities for managing BECDockArea profiles stored in INI files.
Policy:
- All created/modified profiles are stored under the BEC settings root: <base_path>/profiles/{default,user}
- Bundled read-only defaults are discovered in BW core states/default and plugin bec_widgets/profiles but never written to.
- Lookup order when reading: user → settings default → app or plugin bundled default.
- All created/modified profiles are stored under the BEC settings root:
<base_path>/profiles/{baseline,runtime}
- Bundled read-only baselines are discovered in BW core profiles and plugin
bec_widgets/profiles but never written to.
- Lookup order when reading: runtime → settings baseline → app or plugin bundled baseline.
- Legacy settings paths profiles/{default,user} are read through a thin segment
alias layer and copied to the canonical location on first access.
"""
from __future__ import annotations
@@ -32,6 +36,12 @@ logger = bec_logger.logger
MODULE_PATH = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
ProfileOrigin = Literal["module", "plugin", "settings", "unknown"]
ProfileSegment = Literal["baseline", "runtime"]
_PROFILE_SEGMENT_ALIASES: dict[ProfileSegment, tuple[str, str]] = {
"baseline": ("baseline", "default"),
"runtime": ("runtime", "user"),
}
def module_profiles_dir() -> str:
@@ -130,7 +140,7 @@ def _profiles_dir(segment: str, namespace: str | None) -> str:
Build (and ensure) the directory that holds profiles for a namespace segment.
Args:
segment (str): Either ``"user"`` or ``"default"``.
segment (str): Profile segment directory name.
namespace (str | None): Optional namespace label to scope profiles.
Returns:
@@ -143,157 +153,175 @@ def _profiles_dir(segment: str, namespace: str | None) -> str:
return path
def _user_path_candidates(name: str, namespace: str | None) -> list[str]:
"""
Generate candidate user-profile paths honoring namespace fallbacks.
Args:
name (str): Profile name without extension.
namespace (str | None): Optional namespace label.
Returns:
list[str]: Ordered list of candidate user profile paths (.ini files).
"""
def _candidate_namespaces(namespace: str | None) -> list[str | None]:
ns = slugify.slugify(namespace, separator="_") if namespace else None
primary = os.path.join(_profiles_dir("user", ns), f"{name}.ini")
if not ns:
return [primary]
legacy = os.path.join(_profiles_dir("user", None), f"{name}.ini")
return [primary, legacy] if legacy != primary else [primary]
return [None]
return [ns, None]
def _default_path_candidates(name: str, namespace: str | None) -> list[str]:
def _segment_profile_path(segment_name: str, name: str, namespace: str | None) -> str:
return os.path.join(_profiles_dir(segment_name, namespace), f"{name}.ini")
def _canonical_profile_path(segment: ProfileSegment, name: str, namespace: str | None) -> str:
return _segment_profile_path(_PROFILE_SEGMENT_ALIASES[segment][0], name, namespace)
def _segment_path_candidates(
segment: ProfileSegment,
name: str,
namespace: str | None,
*,
include_legacy: bool = True,
migrate_legacy: bool = True,
) -> list[str]:
"""
Generate candidate default-profile paths honoring namespace fallbacks.
Generate profile candidates for a canonical segment.
Args:
name (str): Profile name without extension.
namespace (str | None): Optional namespace label.
Returns:
list[str]: Ordered list of candidate default profile paths (.ini files).
Canonical baseline/runtime files are always preferred. Namespace fallback
files and legacy default/user files are copied to the primary canonical path
when the primary file does not exist.
"""
ns = slugify.slugify(namespace, separator="_") if namespace else None
primary = os.path.join(_profiles_dir("default", ns), f"{name}.ini")
if not ns:
return [primary]
legacy = os.path.join(_profiles_dir("default", None), f"{name}.ini")
return [primary, legacy] if legacy != primary else [primary]
canonical = [
_segment_profile_path(_PROFILE_SEGMENT_ALIASES[segment][0], name, ns)
for ns in _candidate_namespaces(namespace)
]
legacy = []
if include_legacy:
legacy = [
_segment_profile_path(_PROFILE_SEGMENT_ALIASES[segment][1], name, ns)
for ns in _candidate_namespaces(namespace)
]
primary_canonical = canonical[0]
if migrate_legacy and not os.path.exists(primary_canonical):
canonical_src = next((path for path in canonical[1:] if os.path.exists(path)), None)
if canonical_src:
os.makedirs(os.path.dirname(primary_canonical), exist_ok=True)
shutil.copy2(canonical_src, primary_canonical)
elif include_legacy:
legacy_src = next((path for path in legacy if os.path.exists(path)), None)
if legacy_src:
os.makedirs(os.path.dirname(primary_canonical), exist_ok=True)
shutil.copy2(legacy_src, primary_canonical)
return list(dict.fromkeys(canonical + legacy))
def default_profiles_dir(namespace: str | None = None) -> str:
def baseline_profiles_dir(namespace: str | None = None) -> str:
"""
Return the directory that stores default profiles for the namespace.
Return the directory that stores baseline profiles for the namespace.
Args:
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
str: Absolute path to the default profile directory.
str: Absolute path to the baseline profile directory.
"""
return _profiles_dir("default", namespace)
return _profiles_dir("baseline", namespace)
def user_profiles_dir(namespace: str | None = None) -> str:
def runtime_profiles_dir(namespace: str | None = None) -> str:
"""
Return the directory that stores user profiles for the namespace.
Return the directory that stores runtime profiles for the namespace.
Args:
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
str: Absolute path to the user profile directory.
str: Absolute path to the runtime profile directory.
"""
return _profiles_dir("user", namespace)
return _profiles_dir("runtime", namespace)
def default_profile_path(name: str, namespace: str | None = None) -> str:
def baseline_profile_path(name: str, namespace: str | None = None) -> str:
"""
Compute the canonical default profile path for a profile name.
Compute the canonical baseline profile path for a profile name.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
str: Absolute path to the default profile file (.ini).
str: Absolute path to the baseline profile file (.ini).
"""
return _default_path_candidates(name, namespace)[0]
return _canonical_profile_path("baseline", name, namespace)
def user_profile_path(name: str, namespace: str | None = None) -> str:
def runtime_profile_path(name: str, namespace: str | None = None) -> str:
"""
Compute the canonical user profile path for a profile name.
Compute the canonical runtime profile path for a profile name.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
str: Absolute path to the user profile file (.ini).
str: Absolute path to the runtime profile file (.ini).
"""
return _user_path_candidates(name, namespace)[0]
return _canonical_profile_path("runtime", name, namespace)
def user_profile_candidates(name: str, namespace: str | None = None) -> list[str]:
def runtime_profile_candidates(name: str, namespace: str | None = None) -> list[str]:
"""
List all user profile path candidates for a profile name.
List all runtime profile path candidates for a profile name.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
list[str]: De-duplicated list of candidate user profile paths.
list[str]: De-duplicated list of candidate runtime profile paths.
"""
return list(dict.fromkeys(_user_path_candidates(name, namespace)))
return _segment_path_candidates("runtime", name, namespace)
def default_profile_candidates(name: str, namespace: str | None = None) -> list[str]:
def baseline_profile_candidates(name: str, namespace: str | None = None) -> list[str]:
"""
List all default profile path candidates for a profile name.
List all baseline profile path candidates for a profile name.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
list[str]: De-duplicated list of candidate default profile paths.
list[str]: De-duplicated list of candidate baseline profile paths.
"""
return list(dict.fromkeys(_default_path_candidates(name, namespace)))
return _segment_path_candidates("baseline", name, namespace)
def _existing_user_settings(name: str, namespace: str | None = None) -> QSettings | None:
def _existing_runtime_settings(name: str, namespace: str | None = None) -> QSettings | None:
"""
Resolve the first existing user profile settings object.
Resolve the first existing runtime profile settings object.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label to search. Defaults to ``None``.
Returns:
QSettings | None: Config for the first existing user profile candidate, or ``None``
QSettings | None: Config for the first existing runtime profile candidate, or ``None``
when no files are present.
"""
for path in user_profile_candidates(name, namespace):
for path in runtime_profile_candidates(name, namespace):
if os.path.exists(path):
return QSettings(path, QSettings.IniFormat)
return None
def _existing_default_settings(name: str, namespace: str | None = None) -> QSettings | None:
def _existing_baseline_settings(name: str, namespace: str | None = None) -> QSettings | None:
"""
Resolve the first existing default profile settings object.
Resolve the first existing baseline profile settings object.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label to search. Defaults to ``None``.
Returns:
QSettings | None: Config for the first existing default profile candidate, or ``None``
QSettings | None: Config for the first existing baseline profile candidate, or ``None``
when no files are present.
"""
for path in default_profile_candidates(name, namespace):
for path in baseline_profile_candidates(name, namespace):
if os.path.exists(path):
return QSettings(path, QSettings.IniFormat)
return None
@@ -347,7 +375,7 @@ def profile_origin(name: str, namespace: str | None = None) -> ProfileOrigin:
plugin_path = plugin_profile_path(name)
if plugin_path and os.path.exists(plugin_path):
return "plugin"
for path in user_profile_candidates(name, namespace) + default_profile_candidates(
for path in runtime_profile_candidates(name, namespace) + baseline_profile_candidates(
name, namespace
):
if os.path.exists(path):
@@ -406,8 +434,8 @@ def delete_profile_files(name: str, namespace: str | None = None) -> bool:
read_only = is_profile_read_only(name, namespace)
removed = False
# Always allow removing user copies; keep default copies for read-only origins.
for path in set(user_profile_candidates(name, namespace)):
# Always allow removing runtime copies; keep baseline copies for read-only origins.
for path in set(runtime_profile_candidates(name, namespace)):
try:
os.remove(path)
removed = True
@@ -415,7 +443,7 @@ def delete_profile_files(name: str, namespace: str | None = None) -> bool:
continue
if not read_only:
for path in set(default_profile_candidates(name, namespace)):
for path in set(baseline_profile_candidates(name, namespace)):
try:
os.remove(path)
removed = True
@@ -443,7 +471,7 @@ SETTINGS_KEYS = {
def list_profiles(namespace: str | None = None) -> list[str]:
"""
Enumerate all known profile names, syncing bundled defaults when missing locally.
Enumerate all known profile names, syncing bundled baselines when missing locally.
Args:
namespace (str | None, optional): Namespace label scoped to the profile set.
@@ -459,16 +487,27 @@ def list_profiles(namespace: str | None = None) -> list[str]:
return set()
return {os.path.splitext(f)[0] for f in os.listdir(directory) if f.endswith(".ini")}
settings_dirs = {default_profiles_dir(namespace), user_profiles_dir(namespace)}
settings_dirs = {baseline_profiles_dir(namespace), runtime_profiles_dir(namespace)}
if ns:
settings_dirs.add(default_profiles_dir(None))
settings_dirs.add(user_profiles_dir(None))
settings_dirs.add(baseline_profiles_dir(None))
settings_dirs.add(runtime_profiles_dir(None))
for segment in ("baseline", "runtime"):
for legacy_dir in [
_profiles_dir(_PROFILE_SEGMENT_ALIASES[segment][1], item)
for item in _candidate_namespaces(namespace)
]:
settings_dirs.add(legacy_dir)
settings_names: set[str] = set()
for directory in settings_dirs:
settings_names |= _collect_from(directory)
# Also consider read-only defaults from core module and beamline plugin repositories
for name in sorted(settings_names):
runtime_profile_candidates(name, namespace)
baseline_profile_candidates(name, namespace)
# Also consider read-only baselines from core module and beamline plugin repositories
read_only_sources: dict[str, tuple[str, str]] = {}
sources: list[tuple[str, str | None]] = [
("module", module_profiles_dir()),
@@ -484,17 +523,17 @@ def list_profiles(namespace: str | None = None) -> list[str]:
read_only_sources.setdefault(name, (origin, os.path.join(directory, filename)))
for name, (_origin, src) in sorted(read_only_sources.items()):
# Ensure a copy in the namespace-specific settings default directory
dst_default = default_profile_path(name, namespace)
if not os.path.exists(dst_default):
os.makedirs(os.path.dirname(dst_default), exist_ok=True)
shutil.copyfile(src, dst_default)
# Ensure a user copy exists to allow edits in the writable settings area
dst_user = user_profile_path(name, namespace)
if not os.path.exists(dst_user):
os.makedirs(os.path.dirname(dst_user), exist_ok=True)
shutil.copyfile(src, dst_user)
s = open_user_settings(name, namespace)
# Ensure a copy in the namespace-specific settings baseline directory.
dst_baseline = baseline_profile_path(name, namespace)
if not os.path.exists(dst_baseline):
os.makedirs(os.path.dirname(dst_baseline), exist_ok=True)
shutil.copy2(src, dst_baseline)
# Ensure a runtime copy exists to allow edits in the writable settings area.
dst_runtime = runtime_profile_path(name, namespace)
if not os.path.exists(dst_runtime):
os.makedirs(os.path.dirname(dst_runtime), exist_ok=True)
shutil.copy2(src, dst_runtime)
s = open_runtime_settings(name, namespace)
if s.value(SETTINGS_KEYS["created_at"], "") == "":
s.setValue(SETTINGS_KEYS["created_at"], now_iso_utc())
@@ -504,32 +543,34 @@ def list_profiles(namespace: str | None = None) -> list[str]:
return sorted(settings_names)
def open_default_settings(name: str, namespace: str | None = None) -> QSettings:
def open_baseline_settings(name: str, namespace: str | None = None) -> QSettings:
"""
Open (and create if necessary) the default profile settings file.
Open (and create if necessary) the baseline profile settings file.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
QSettings: Settings instance targeting the default profile file.
QSettings: Settings instance targeting the baseline profile file.
"""
return QSettings(default_profile_path(name, namespace), QSettings.IniFormat)
baseline_profile_candidates(name, namespace)
return QSettings(baseline_profile_path(name, namespace), QSettings.IniFormat)
def open_user_settings(name: str, namespace: str | None = None) -> QSettings:
def open_runtime_settings(name: str, namespace: str | None = None) -> QSettings:
"""
Open (and create if necessary) the user profile settings file.
Open (and create if necessary) the runtime profile settings file.
Args:
name (str): Profile name without extension.
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
QSettings: Settings instance targeting the user profile file.
QSettings: Settings instance targeting the runtime profile file.
"""
return QSettings(user_profile_path(name, namespace), QSettings.IniFormat)
runtime_profile_candidates(name, namespace)
return QSettings(runtime_profile_path(name, namespace), QSettings.IniFormat)
def _app_settings() -> QSettings:
@@ -759,26 +800,26 @@ def read_manifest(settings: QSettings) -> list[dict]:
return items
def restore_user_from_default(name: str, namespace: str | None = None) -> None:
def restore_runtime_from_baseline(name: str, namespace: str | None = None) -> None:
"""
Copy the default profile to the user profile, preserving quick-select flag.
Copy the baseline profile to the runtime profile, preserving quick-select flag.
Args:
name(str): Profile name without extension.
namespace(str | None, optional): Namespace label. Defaults to ``None``.
"""
src = None
for candidate in default_profile_candidates(name, namespace):
for candidate in baseline_profile_candidates(name, namespace):
if os.path.exists(candidate):
src = candidate
break
if not src:
return
dst = user_profile_path(name, namespace)
dst = runtime_profile_path(name, namespace)
preserve_quick_select = is_quick_select(name, namespace)
os.makedirs(os.path.dirname(dst), exist_ok=True)
shutil.copyfile(src, dst)
s = open_user_settings(name, namespace)
s = open_runtime_settings(name, namespace)
if not s.value(SETTINGS_KEYS["created_at"], ""):
s.setValue(SETTINGS_KEYS["created_at"], now_iso_utc())
if preserve_quick_select:
@@ -796,9 +837,9 @@ def is_quick_select(name: str, namespace: str | None = None) -> bool:
Returns:
bool: True if quick-select is enabled for the profile.
"""
s = _existing_user_settings(name, namespace)
s = _existing_runtime_settings(name, namespace)
if s is None:
s = _existing_default_settings(name, namespace)
s = _existing_baseline_settings(name, namespace)
if s is None:
return False
return s.value(SETTINGS_KEYS["is_quick_select"], False, type=bool)
@@ -813,13 +854,13 @@ def set_quick_select(name: str, enabled: bool, namespace: str | None = None) ->
enabled(bool): True to enable quick-select, False to disable.
namespace(str | None, optional): Namespace label. Defaults to ``None``.
"""
s = open_user_settings(name, namespace)
s = open_runtime_settings(name, namespace)
s.setValue(SETTINGS_KEYS["is_quick_select"], bool(enabled))
def list_quick_profiles(namespace: str | None = None) -> list[str]:
"""
List only profiles that have quick-select enabled (user wins over default).
List only profiles that have quick-select enabled (runtime wins over baseline).
Args:
namespace(str | None, optional): Namespace label. Defaults to ``None``.
@@ -909,8 +950,8 @@ class ProfileInfo(BaseModel):
is_quick_select: bool = False
widget_count: int = 0
size_kb: int = 0
user_path: str = ""
default_path: str = ""
runtime_path: str = ""
baseline_path: str = ""
origin: ProfileOrigin = "unknown"
is_read_only: bool = False
@@ -924,19 +965,19 @@ def get_profile_info(name: str, namespace: str | None = None) -> ProfileInfo:
namespace (str | None, optional): Namespace label. Defaults to ``None``.
Returns:
ProfileInfo: Structured profile metadata, preferring the user copy when present.
ProfileInfo: Structured profile metadata, preferring the runtime copy when present.
"""
user_paths = user_profile_candidates(name, namespace)
default_paths = default_profile_candidates(name, namespace)
u_path = next((p for p in user_paths if os.path.exists(p)), user_paths[0])
d_path = next((p for p in default_paths if os.path.exists(p)), default_paths[0])
runtime_paths = runtime_profile_candidates(name, namespace)
baseline_paths = baseline_profile_candidates(name, namespace)
r_path = next((p for p in runtime_paths if os.path.exists(p)), runtime_paths[0])
b_path = next((p for p in baseline_paths if os.path.exists(p)), baseline_paths[0])
origin = profile_origin(name, namespace)
read_only = origin in {"module", "plugin"}
prefer_user = os.path.exists(u_path)
if prefer_user:
s = QSettings(u_path, QSettings.IniFormat)
elif os.path.exists(d_path):
s = QSettings(d_path, QSettings.IniFormat)
prefer_runtime = os.path.exists(r_path)
if prefer_runtime:
s = QSettings(r_path, QSettings.IniFormat)
elif os.path.exists(b_path):
s = QSettings(b_path, QSettings.IniFormat)
else:
s = None
if s is None:
@@ -957,14 +998,14 @@ def get_profile_info(name: str, namespace: str | None = None) -> ProfileInfo:
is_quick_select=False,
widget_count=0,
size_kb=0,
user_path=u_path,
default_path=d_path,
runtime_path=r_path,
baseline_path=b_path,
origin=origin,
is_read_only=read_only,
)
created = s.value(SETTINGS_KEYS["created_at"], "", type=str) or now_iso_utc()
src_path = u_path if prefer_user else d_path
src_path = r_path if prefer_runtime else b_path
modified = _file_modified_iso(src_path)
count = _manifest_count(s)
try:
@@ -990,8 +1031,8 @@ def get_profile_info(name: str, namespace: str | None = None) -> ProfileInfo:
is_quick_select=is_quick_select(name, namespace),
widget_count=count,
size_kb=size_kb,
user_path=u_path,
default_path=d_path,
runtime_path=r_path,
baseline_path=b_path,
origin=origin,
is_read_only=read_only,
)
@@ -999,7 +1040,7 @@ def get_profile_info(name: str, namespace: str | None = None) -> ProfileInfo:
def load_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap | None:
"""
Load the stored screenshot pixmap for a profile from settings (user preferred).
Load the stored screenshot pixmap for a profile from settings (runtime preferred).
Args:
name (str): Profile name without extension.
@@ -1008,17 +1049,17 @@ def load_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap
Returns:
QPixmap | None: Screenshot pixmap or ``None`` if unavailable.
"""
s = _existing_user_settings(name, namespace)
s = _existing_runtime_settings(name, namespace)
if s is None:
s = _existing_default_settings(name, namespace)
s = _existing_baseline_settings(name, namespace)
if s is None:
return None
return _load_screenshot_from_settings(s)
def load_default_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap | None:
def load_baseline_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap | None:
"""
Load the screenshot from the default profile copy, if available.
Load the screenshot from the baseline profile copy, if available.
Args:
name (str): Profile name without extension.
@@ -1027,15 +1068,15 @@ def load_default_profile_screenshot(name: str, namespace: str | None = None) ->
Returns:
QPixmap | None: Screenshot pixmap or ``None`` if unavailable.
"""
s = _existing_default_settings(name, namespace)
s = _existing_baseline_settings(name, namespace)
if s is None:
return None
return _load_screenshot_from_settings(s)
def load_user_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap | None:
def load_runtime_profile_screenshot(name: str, namespace: str | None = None) -> QPixmap | None:
"""
Load the screenshot from the user profile copy, if available.
Load the screenshot from the runtime profile copy, if available.
Args:
name (str): Profile name without extension.
@@ -1044,7 +1085,7 @@ def load_user_profile_screenshot(name: str, namespace: str | None = None) -> QPi
Returns:
QPixmap | None: Screenshot pixmap or ``None`` if unavailable.
"""
s = _existing_user_settings(name, namespace)
s = _existing_runtime_settings(name, namespace)
if s is None:
return None
return _load_screenshot_from_settings(s)
@@ -160,7 +160,7 @@ class SaveProfileDialog(QDialog):
self,
"Read-only profile",
(
f"'{name}' is a default profile provided by {provider} and cannot be overwritten.\n"
f"'{name}' is a baseline profile provided by {provider} and cannot be overwritten.\n"
"Please choose a different name."
),
)
@@ -179,7 +179,7 @@ class SaveProfileDialog(QDialog):
"Overwrite profile",
(
f"A profile named '{name}' already exists.\n\n"
"Overwriting will update both the saved profile and its restore default.\n"
"Overwriting will update both the runtime profile and its restore baseline.\n"
"Do you want to continue?"
),
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
@@ -257,21 +257,24 @@ class PreviewPanel(QGroupBox):
class RestoreProfileDialog(QDialog):
"""
Confirmation dialog that previews the current profile screenshot against the default baseline.
Confirmation dialog that previews the current runtime screenshot against the baseline.
"""
def __init__(
self, parent: QWidget | None, current_pixmap: QPixmap | None, default_pixmap: QPixmap | None
self,
parent: QWidget | None,
current_pixmap: QPixmap | None,
baseline_pixmap: QPixmap | None,
):
super().__init__(parent)
self.setWindowTitle("Restore Profile to Default")
self.setWindowTitle("Restore Profile to Baseline")
self.setModal(True)
self.resize(880, 480)
layout = QVBoxLayout(self)
info_label = QLabel(
"Restoring will discard your custom layout and replace it with the default profile."
"Restoring will discard your runtime layout and replace it with the baseline profile."
)
info_label.setWordWrap(True)
layout.addWidget(info_label)
@@ -280,7 +283,7 @@ class RestoreProfileDialog(QDialog):
layout.addLayout(preview_row)
current_preview = PreviewPanel("Current", current_pixmap, self)
default_preview = PreviewPanel("Default", default_pixmap, self)
baseline_preview = PreviewPanel("Baseline", baseline_pixmap, self)
# Equal expansion left/right
preview_row.addWidget(current_preview, 1)
@@ -292,7 +295,7 @@ class RestoreProfileDialog(QDialog):
arrow_label.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Expanding)
preview_row.addWidget(arrow_label)
preview_row.addWidget(default_preview, 1)
preview_row.addWidget(baseline_preview, 1)
# Enforce equal stretch for both previews
preview_row.setStretch(0, 1)
@@ -300,7 +303,7 @@ class RestoreProfileDialog(QDialog):
preview_row.setStretch(2, 1)
warn_label = QLabel(
"This action cannot be undone. Do you want to restore the default layout now?"
"This action cannot be undone. Do you want to restore the baseline layout now?"
)
warn_label.setWordWrap(True)
layout.addWidget(warn_label)
@@ -324,7 +327,7 @@ class RestoreProfileDialog(QDialog):
@staticmethod
def confirm(
parent: QWidget | None, current_pixmap: QPixmap | None, default_pixmap: QPixmap | None
parent: QWidget | None, current_pixmap: QPixmap | None, baseline_pixmap: QPixmap | None
) -> bool:
dialog = RestoreProfileDialog(parent, current_pixmap, default_pixmap)
dialog = RestoreProfileDialog(parent, current_pixmap, baseline_pixmap)
return dialog.exec() == QDialog.Accepted
@@ -48,7 +48,7 @@ class WorkSpaceManager(BECWidget, QWidget):
HEADERS = ["Actions", "Profile", "Author"]
def __init__(
self, parent=None, target_widget=None, default_profile: str | None = None, **kwargs
self, parent=None, target_widget=None, active_profile: str | None = None, **kwargs
):
super().__init__(parent=parent, **kwargs)
self.target_widget = target_widget
@@ -59,13 +59,13 @@ class WorkSpaceManager(BECWidget, QWidget):
self._init_ui()
if self.target_widget is not None and hasattr(self.target_widget, "profile_changed"):
self.target_widget.profile_changed.connect(self.on_profile_changed)
if default_profile is not None:
self._select_by_name(default_profile)
self._show_profile_details(default_profile)
if active_profile is not None:
self._select_by_name(active_profile)
self._show_profile_details(active_profile)
def _init_ui(self):
self.root_layout = QHBoxLayout(self)
self.splitter = QSplitter(Qt.Horizontal, self)
self.splitter = QSplitter(Qt.Orientation.Horizontal, self)
self.root_layout.addWidget(self.splitter)
# Init components
@@ -89,7 +89,9 @@ class WorkSpaceManager(BECWidget, QWidget):
left_panel.setMinimumWidth(220)
# Make the screenshot preview expand to fill remaining space
self.screenshot_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
self.screenshot_label.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding
)
self.right_box = QGroupBox("Profile Screenshot Preview", self)
right_col = QVBoxLayout(self.right_box)
@@ -250,8 +252,8 @@ class WorkSpaceManager(BECWidget, QWidget):
("Quick select", "Yes" if info.is_quick_select else "No"),
("Widgets", str(info.widget_count)),
("Size (KB)", str(info.size_kb)),
("User path", info.user_path or ""),
("Default path", info.default_path or ""),
("Runtime path", info.runtime_path or ""),
("Baseline path", info.baseline_path or ""),
]
for k, v in entries:
self.profile_details_tree.addTopLevelItem(QTreeWidgetItem([k, v]))
@@ -24,19 +24,9 @@ class ProfileComboBox(QComboBox):
def set_quick_profile_provider(self, provider: Callable[[], list[str]]) -> None:
self._quick_provider = provider
def refresh_profiles(
self, active_profile: str | None = None, show_empty_profile: bool = False
def _refresh_profiles(
self, current_text: str, active_profile: str | None = None, show_empty_profile: bool = False
) -> None:
"""
Refresh the profile list and ensure the active profile is visible.
Args:
active_profile(str | None): The currently active profile name.
show_empty_profile(bool): If True, show an explicit empty unsaved workspace entry.
"""
current_text = active_profile or self.currentText()
self.blockSignals(True)
self.clear()
quick_profiles = self._quick_provider()
@@ -103,7 +93,6 @@ class ProfileComboBox(QComboBox):
if index >= 0:
self.setCurrentIndex(index)
self.blockSignals(False)
if active_profile and self.currentText() != active_profile:
idx = self.findText(active_profile)
if idx >= 0:
@@ -115,6 +104,24 @@ class ProfileComboBox(QComboBox):
else:
self.setToolTip("")
def refresh_profiles(
self, active_profile: str | None = None, show_empty_profile: bool = False
) -> None:
"""
Refresh the profile list and ensure the active profile is visible.
Args:
active_profile(str | None): The currently active profile name.
show_empty_profile(bool): If True, show an explicit empty unsaved workspace entry.
"""
current_text = active_profile or self.currentText()
was_blocked = self.blockSignals(True)
try:
self._refresh_profiles(current_text, active_profile, show_empty_profile)
finally:
self.blockSignals(was_blocked)
def workspace_bundle(components: ToolbarComponents, enable_tools: bool = True) -> ToolbarBundle:
"""
@@ -122,6 +129,7 @@ def workspace_bundle(components: ToolbarComponents, enable_tools: bool = True) -
Args:
components (ToolbarComponents): The components to be added to the bundle.
enable_tools(bool): If True, show the workspace management tools; otherwise, only show the profile combo.
Returns:
ToolbarBundle: The workspace toolbar bundle.
@@ -143,15 +151,15 @@ def workspace_bundle(components: ToolbarComponents, enable_tools: bool = True) -
components.get_action("save_workspace").action.setVisible(enable_tools)
components.add_safe(
"reset_default_workspace",
"reset_baseline_workspace",
MaterialIconAction(
icon_name="undo",
tooltip="Refresh Current Workspace",
tooltip="Restore Baseline Profile",
checkable=False,
parent=components.toolbar,
),
)
components.get_action("reset_default_workspace").action.setVisible(enable_tools)
components.get_action("reset_baseline_workspace").action.setVisible(enable_tools)
components.add_safe(
"manage_workspaces",
@@ -164,7 +172,7 @@ def workspace_bundle(components: ToolbarComponents, enable_tools: bool = True) -
bundle = ToolbarBundle("workspace", components)
bundle.add_action("workspace_combo")
bundle.add_action("save_workspace")
bundle.add_action("reset_default_workspace")
bundle.add_action("reset_baseline_workspace")
bundle.add_action("manage_workspaces")
return bundle
@@ -194,9 +202,9 @@ class WorkspaceConnection(BundleConnection):
self.target_widget.load_profile
)
reset_action = self.components.get_action("reset_default_workspace").action
reset_action = self.components.get_action("reset_baseline_workspace").action
if reset_action.isVisible():
reset_action.triggered.connect(self._reset_workspace_to_default)
reset_action.triggered.connect(self._reset_workspace_to_baseline)
manage_action = self.components.get_action("manage_workspaces").action
if manage_action.isVisible():
@@ -213,9 +221,9 @@ class WorkspaceConnection(BundleConnection):
self.target_widget.load_profile
)
reset_action = self.components.get_action("reset_default_workspace").action
reset_action = self.components.get_action("reset_baseline_workspace").action
if reset_action.isVisible():
reset_action.triggered.disconnect(self._reset_workspace_to_default)
reset_action.triggered.disconnect(self._reset_workspace_to_baseline)
manage_action = self.components.get_action("manage_workspaces").action
if manage_action.isVisible():
@@ -223,8 +231,8 @@ class WorkspaceConnection(BundleConnection):
self._connected = False
@SafeSlot()
def _reset_workspace_to_default(self):
def _reset_workspace_to_baseline(self):
"""
Refreshes the current workspace.
"""
self.target_widget.restore_user_profile_from_default()
self.target_widget.restore_baseline_profile(show_dialog=True)
@@ -4,11 +4,7 @@ import webbrowser
class BECWebLinksMixin:
@staticmethod
def open_bec_docs():
webbrowser.open("https://beamline-experiment-control.readthedocs.io/en/latest/")
@staticmethod
def open_bec_widgets_docs():
webbrowser.open("https://bec.readthedocs.io/projects/bec-widgets/en/latest/")
webbrowser.open("https://bec-project.github.io/bec_docs/")
@staticmethod
def open_bec_bug_report():
@@ -355,17 +355,13 @@ class BECMainWindow(BECWidget, QMainWindow):
bec_docs = QAction("BEC Docs", self)
bec_docs.setIcon(help_icon)
widgets_docs = QAction("BEC Widgets Docs", self)
widgets_docs.setIcon(help_icon)
bug_report = QAction("Bug Report", self)
bug_report.setIcon(bug_icon)
bec_docs.triggered.connect(BECWebLinksMixin.open_bec_docs)
widgets_docs.triggered.connect(BECWebLinksMixin.open_bec_widgets_docs)
bug_report.triggered.connect(BECWebLinksMixin.open_bec_bug_report)
help_menu.addAction(bec_docs)
help_menu.addAction(widgets_docs)
help_menu.addAction(bug_report)
################################################################################
+1 -1
View File
@@ -611,7 +611,7 @@ class Heatmap(ImageBase):
scan_msg = self.scan_item.status_message
elif hasattr(self.scan_item, "metadata"):
metadata = self.scan_item.metadata["bec"]
status = metadata["exit_status"]
status = metadata["status"]
scan_id = metadata["scan_id"]
scan_name = metadata["scan_name"]
scan_type = metadata["scan_type"]
+2 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "bec_widgets"
version = "3.7.0"
version = "3.8.0"
description = "BEC Widgets"
requires-python = ">=3.11"
classifiers = [
@@ -52,6 +52,7 @@ dev = [
"pytest-xvfb~=3.0",
"pytest~=8.0",
"pytest-cov~=6.1.1",
"pytest-benchmark~=5.2",
"watchdog~=6.0",
"pre_commit~=4.2",
]
@@ -0,0 +1,5 @@
#!/usr/bin/env bash
# BENCHMARK_TITLE: Import bec_widgets
set -euo pipefail
python -c 'import bec_widgets; print(bec_widgets.__file__)'
@@ -0,0 +1,5 @@
#!/usr/bin/env bash
# BENCHMARK_TITLE: BEC IPython client with companion app
set -euo pipefail
bec --post-startup-file tests/benchmarks/hyperfine/utils/exit_bec_startup.py
@@ -0,0 +1,5 @@
#!/usr/bin/env bash
# BENCHMARK_TITLE: BEC IPython client without companion app
set -euo pipefail
bec --nogui --post-startup-file tests/benchmarks/hyperfine/utils/exit_bec_startup.py
@@ -0,0 +1,5 @@
import time
_ip = get_ipython()
_ip.confirm_exit = False
_ip.ask_exit()
-1
View File
@@ -59,5 +59,4 @@ def test_run_line_scan_with_parameters_e2e(scan_control, bec_client_lib, qtbot):
last_scan = queue.scan_storage.storage[-1]
assert last_scan.status_message.info["scan_name"] == scan_name
assert last_scan.status_message.info["exp_time"] == kwargs["exp_time"]
assert last_scan.status_message.info["scan_motors"] == [args["device"]]
assert last_scan.status_message.info["num_points"] == kwargs["steps"]
-1
View File
@@ -84,7 +84,6 @@ def test_scan_metadata_for_custom_scan(
last_scan = queue.scan_storage.storage[-1]
assert last_scan.status_message.info["scan_name"] == scan_name
assert last_scan.status_message.info["exp_time"] == kwargs["exp_time"]
assert last_scan.status_message.info["scan_motors"] == [args["device"]]
assert last_scan.status_message.info["num_points"] == kwargs["steps"]
if valid:
@@ -0,0 +1,27 @@
from __future__ import annotations
import pytest
from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from tests.unit_tests.client_mocks import mocked_client
@pytest.fixture
def dock_area(qtbot, mocked_client):
widget = BECDockArea(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_add_waveform_to_dock_area(benchmark, dock_area, qtbot, mocked_client):
"""Benchmark adding a Waveform widget to an existing dock area."""
def add_waveform():
dock_area.new("Waveform")
return dock_area
dock = benchmark(add_waveform)
assert dock is not None
+3 -3
View File
@@ -227,7 +227,7 @@ def create_history_file(file_path, data: dict, metadata: dict) -> messages.ScanH
msg = messages.ScanHistoryMessage(
scan_id=metadata["scan_id"],
scan_name=metadata["scan_name"],
exit_status=metadata["exit_status"],
exit_status=metadata["status"],
file_path=file_path,
scan_number=metadata["scan_number"],
dataset_number=metadata["dataset_number"],
@@ -274,7 +274,7 @@ def grid_scan_history_msg(tmpdir):
"scan_id": "test_scan",
"scan_name": "grid_scan",
"scan_type": "step",
"exit_status": "closed",
"status": "closed",
"scan_number": 1,
"dataset_number": 1,
"request_inputs": {
@@ -354,7 +354,7 @@ def scan_history_factory(tmpdir):
"scan_id": scan_id,
"scan_name": scan_name,
"scan_type": scan_type,
"exit_status": "closed",
"status": "closed",
"scan_number": scan_number,
"dataset_number": dataset_number,
"request_inputs": {
-1
View File
@@ -71,7 +71,6 @@ def bec_queue_msg_full():
},
"report_instructions": [{"scan_progress": 20}],
"scan_id": "2d704cc3-c172-404c-866d-608ce09fce40",
"scan_motors": ["samx"],
"scan_number": 1289,
}
],
+2 -5
View File
@@ -146,10 +146,7 @@ class TestDeviceManagerViewDialogs:
group_combo: QtWidgets.QComboBox = dialog._control_widgets["group_combo"]
assert group_combo.count() == len(OPHYD_DEVICE_TEMPLATES)
# Test select a group from available templates
variant_combo = dialog._control_widgets["variant_combo"]
assert variant_combo.isEnabled() is False
with qtbot.waitSignal(group_combo.currentTextChanged):
epics_signal_index = group_combo.findText("EpicsSignal")
group_combo.setCurrentIndex(epics_signal_index) # Select "EpicsSignal" group
@@ -235,7 +232,7 @@ class TestDeviceManagerViewDialogs:
sample_config = {
"name": "TestDevice",
"enabled": True,
"deviceClass": "ophyd.EpicsSignal",
"deviceClass": "ophyd_devices.EpicsSignal",
"readoutPriority": "baseline",
"deviceConfig": {"read_pv": "X25DA-ES1-MOT:GET"},
}
@@ -248,7 +245,7 @@ class TestDeviceManagerViewDialogs:
assert variant_combo.currentText() == "EpicsSignal"
config = dialog._device_config_template.get_config_fields()
assert config["name"] == "TestDevice"
assert config["deviceClass"] == "ophyd.EpicsSignal"
assert config["deviceClass"] == "ophyd_devices.EpicsSignal"
assert config["deviceConfig"]["read_pv"] == "X25DA-ES1-MOT:GET"
# Test now to add the device config with different validation results
+312 -180
View File
@@ -19,19 +19,19 @@ from bec_widgets.widgets.containers.dock_area.basic_dock_area import (
from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea, SaveProfileDialog
from bec_widgets.widgets.containers.dock_area.profile_utils import (
SETTINGS_KEYS,
default_profile_path,
baseline_profile_path,
get_profile_info,
is_profile_read_only,
is_quick_select,
list_profiles,
load_default_profile_screenshot,
load_user_profile_screenshot,
open_default_settings,
open_user_settings,
load_baseline_profile_screenshot,
load_runtime_profile_screenshot,
open_baseline_settings,
open_runtime_settings,
read_manifest,
restore_user_from_default,
restore_runtime_from_baseline,
runtime_profile_path,
set_quick_select,
user_profile_path,
write_manifest,
)
from bec_widgets.widgets.containers.dock_area.settings.dialogs import (
@@ -188,17 +188,17 @@ class _NamespaceProfiles:
def __init__(self, widget: BECDockArea):
self.namespace = widget.profile_namespace
def open_user(self, name: str):
return open_user_settings(name, namespace=self.namespace)
def open_runtime(self, name: str):
return open_runtime_settings(name, namespace=self.namespace)
def open_default(self, name: str):
return open_default_settings(name, namespace=self.namespace)
def open_baseline(self, name: str):
return open_baseline_settings(name, namespace=self.namespace)
def user_path(self, name: str) -> str:
return user_profile_path(name, namespace=self.namespace)
def runtime_path(self, name: str) -> str:
return runtime_profile_path(name, namespace=self.namespace)
def default_path(self, name: str) -> str:
return default_profile_path(name, namespace=self.namespace)
def baseline_path(self, name: str) -> str:
return baseline_profile_path(name, namespace=self.namespace)
def list_profiles(self) -> list[str]:
return list_profiles(namespace=self.namespace)
@@ -615,35 +615,6 @@ class TestBasicDockArea:
]
class TestAdvancedDockAreaInit:
"""Test initialization and basic properties."""
def test_init(self, advanced_dock_area):
assert advanced_dock_area is not None
assert isinstance(advanced_dock_area, BECDockArea)
assert advanced_dock_area.mode == "creator"
assert hasattr(advanced_dock_area, "dock_manager")
assert hasattr(advanced_dock_area, "toolbar")
assert hasattr(advanced_dock_area, "dark_mode_button")
assert hasattr(advanced_dock_area, "state_manager")
def test_rpc_and_plugin_flags(self):
assert BECDockArea.RPC is True
assert BECDockArea.PLUGIN is False
def test_user_access_list(self):
expected_methods = [
"new",
"widget_map",
"widget_list",
"workspace_is_locked",
"attach_all",
"delete_all",
]
for method in expected_methods:
assert method in BECDockArea.USER_ACCESS
class TestDockManagement:
"""Test dock creation, management, and manipulation."""
@@ -851,16 +822,6 @@ class TestDeveloperMode:
class TestToolbarFunctionality:
"""Test toolbar setup and functionality."""
def test_toolbar_setup(self, advanced_dock_area):
"""Test toolbar is properly set up."""
assert hasattr(advanced_dock_area, "toolbar")
assert hasattr(advanced_dock_area, "_ACTION_MAPPINGS")
# Check that action mappings are properly set
assert "menu_plots" in advanced_dock_area._ACTION_MAPPINGS
assert "menu_devices" in advanced_dock_area._ACTION_MAPPINGS
assert "menu_utils" in advanced_dock_area._ACTION_MAPPINGS
def test_toolbar_plot_actions(self, advanced_dock_area):
"""Test plot toolbar actions trigger widget creation."""
plot_actions = [
@@ -946,7 +907,7 @@ class TestToolbarFunctionality:
def test_load_profile_restores_floating_dock(self, advanced_dock_area, qtbot):
helper = profile_helper(advanced_dock_area)
settings = helper.open_user("floating_profile")
settings = helper.open_runtime("floating_profile")
settings.clear()
settings.setValue("profile/created_at", "2025-11-23T00:00:00Z")
@@ -1215,18 +1176,6 @@ class TestPreviewPanel:
assert "No preview available" in panel.image_label.text()
class TestRestoreProfileDialog:
"""Test restore dialog confirmation flow."""
def test_confirm_accepts(self, monkeypatch):
monkeypatch.setattr(RestoreProfileDialog, "exec", lambda self: QDialog.Accepted)
assert RestoreProfileDialog.confirm(None, QPixmap(), QPixmap()) is True
def test_confirm_rejects(self, monkeypatch):
monkeypatch.setattr(RestoreProfileDialog, "exec", lambda self: QDialog.Rejected)
assert RestoreProfileDialog.confirm(None, QPixmap(), QPixmap()) is False
class TestProfileInfoAndScreenshots:
"""Tests for profile utilities metadata and screenshot helpers."""
@@ -1246,9 +1195,9 @@ class TestProfileInfoAndScreenshots:
settings.endArray()
settings.sync()
def test_get_profile_info_user_origin(self, temp_profile_dir):
name = "info_user"
settings = open_user_settings(name)
def test_get_profile_info_runtime_origin(self, temp_profile_dir):
name = "info_runtime"
settings = open_runtime_settings(name)
settings.setValue(profile_utils.SETTINGS_KEYS["created_at"], "2023-01-01T00:00:00Z")
settings.setValue("profile/author", "Custom")
set_quick_select(name, True)
@@ -1262,22 +1211,22 @@ class TestProfileInfoAndScreenshots:
assert info.is_quick_select is True
assert info.widget_count == 3
assert info.author == "User"
assert info.user_path.endswith(f"{name}.ini")
assert info.runtime_path.endswith(f"{name}.ini")
assert info.size_kb >= 0
def test_get_profile_info_default_only(self, temp_profile_dir):
name = "info_default"
settings = open_default_settings(name)
def test_get_profile_info_baseline_only(self, temp_profile_dir):
name = "info_baseline"
settings = open_baseline_settings(name)
self._write_manifest(settings, count=1)
user_path = user_profile_path(name)
if os.path.exists(user_path):
os.remove(user_path)
runtime_path = runtime_profile_path(name)
if os.path.exists(runtime_path):
os.remove(runtime_path)
info = get_profile_info(name)
assert info.origin == "settings"
assert info.user_path.endswith(f"{name}.ini")
assert info.baseline_path.endswith(f"{name}.ini")
assert info.widget_count == 1
def test_get_profile_info_module_readonly(self, module_profile_factory):
@@ -1289,10 +1238,10 @@ class TestProfileInfoAndScreenshots:
def test_get_profile_info_unknown_profile(self):
name = "nonexistent_profile"
if os.path.exists(user_profile_path(name)):
os.remove(user_profile_path(name))
if os.path.exists(default_profile_path(name)):
os.remove(default_profile_path(name))
if os.path.exists(runtime_profile_path(name)):
os.remove(runtime_profile_path(name))
if os.path.exists(baseline_profile_path(name)):
os.remove(baseline_profile_path(name))
info = get_profile_info(name)
@@ -1300,29 +1249,29 @@ class TestProfileInfoAndScreenshots:
assert info.is_read_only is False
assert info.widget_count == 0
def test_load_user_profile_screenshot(self, temp_profile_dir):
name = "user_screenshot"
settings = open_user_settings(name)
def test_load_runtime_profile_screenshot(self, temp_profile_dir):
name = "runtime_screenshot"
settings = open_runtime_settings(name)
settings.setValue(profile_utils.SETTINGS_KEYS["screenshot"], self.PNG_BYTES)
settings.sync()
pix = load_user_profile_screenshot(name)
pix = load_runtime_profile_screenshot(name)
assert pix is not None and not pix.isNull()
def test_load_default_profile_screenshot(self, temp_profile_dir):
name = "default_screenshot"
settings = open_default_settings(name)
def test_load_baseline_profile_screenshot(self, temp_profile_dir):
name = "baseline_screenshot"
settings = open_baseline_settings(name)
settings.setValue(profile_utils.SETTINGS_KEYS["screenshot"], self.PNG_BYTES)
settings.sync()
pix = load_default_profile_screenshot(name)
pix = load_baseline_profile_screenshot(name)
assert pix is not None and not pix.isNull()
def test_load_screenshot_from_settings_invalid(self, temp_profile_dir):
name = "invalid_screenshot"
settings = open_user_settings(name)
settings = open_runtime_settings(name)
settings.setValue(profile_utils.SETTINGS_KEYS["screenshot"], "not-an-image")
settings.sync()
@@ -1332,7 +1281,7 @@ class TestProfileInfoAndScreenshots:
def test_load_screenshot_from_settings_bytes(self, temp_profile_dir):
name = "bytes_screenshot"
settings = open_user_settings(name)
settings = open_runtime_settings(name)
settings.setValue(profile_utils.SETTINGS_KEYS["screenshot"], self.PNG_BYTES)
settings.sync()
@@ -1347,7 +1296,7 @@ class TestWorkSpaceManager:
@staticmethod
def _create_profiles(names):
for name in names:
settings = open_user_settings(name)
settings = open_runtime_settings(name)
settings.setValue("meta", "value")
settings.sync()
@@ -1411,7 +1360,7 @@ class TestWorkSpaceManager:
manager.delete_profile(name)
assert not os.path.exists(user_profile_path(name))
assert not os.path.exists(runtime_profile_path(name))
assert target.refresh_calls >= 1
def test_delete_readonly_profile_shows_message(
@@ -1441,21 +1390,23 @@ class TestWorkSpaceManager:
class TestAdvancedDockAreaRestoreAndDialogs:
"""Additional coverage for restore flows and workspace dialogs."""
def test_restore_user_profile_from_default_confirm_true(self, advanced_dock_area, monkeypatch):
def test_restore_runtime_profile_from_baseline_confirm_true(
self, advanced_dock_area, monkeypatch
):
profile_name = "profile_restore_true"
helper = profile_helper(advanced_dock_area)
helper.open_default(profile_name).sync()
helper.open_user(profile_name).sync()
helper.open_baseline(profile_name).sync()
helper.open_runtime(profile_name).sync()
advanced_dock_area._current_profile_name = profile_name
advanced_dock_area.isVisible = lambda: False
pix = QPixmap(8, 8)
pix.fill(Qt.red)
monkeypatch.setattr(
"bec_widgets.widgets.containers.dock_area.dock_area.load_user_profile_screenshot",
"bec_widgets.widgets.containers.dock_area.dock_area.load_runtime_profile_screenshot",
lambda name, namespace=None: pix,
)
monkeypatch.setattr(
"bec_widgets.widgets.containers.dock_area.dock_area.load_default_profile_screenshot",
"bec_widgets.widgets.containers.dock_area.dock_area.load_baseline_profile_screenshot",
lambda name, namespace=None: pix,
)
monkeypatch.setattr(
@@ -1465,12 +1416,12 @@ class TestAdvancedDockAreaRestoreAndDialogs:
with (
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.restore_user_from_default"
"bec_widgets.widgets.containers.dock_area.dock_area.restore_runtime_from_baseline"
) as mock_restore,
patch.object(advanced_dock_area, "delete_all") as mock_delete_all,
patch.object(advanced_dock_area, "load_profile") as mock_load_profile,
):
advanced_dock_area.restore_user_profile_from_default()
advanced_dock_area.restore_baseline_profile(show_dialog=True)
assert mock_restore.call_count == 1
args, kwargs = mock_restore.call_args
@@ -1479,20 +1430,22 @@ class TestAdvancedDockAreaRestoreAndDialogs:
mock_delete_all.assert_called_once()
mock_load_profile.assert_called_once_with(profile_name)
def test_restore_user_profile_from_default_confirm_false(self, advanced_dock_area, monkeypatch):
def test_restore_runtime_profile_from_baseline_confirm_false(
self, advanced_dock_area, monkeypatch
):
profile_name = "profile_restore_false"
helper = profile_helper(advanced_dock_area)
helper.open_default(profile_name).sync()
helper.open_user(profile_name).sync()
helper.open_baseline(profile_name).sync()
helper.open_runtime(profile_name).sync()
advanced_dock_area._current_profile_name = profile_name
advanced_dock_area.isVisible = lambda: False
monkeypatch.setattr(
"bec_widgets.widgets.containers.dock_area.dock_area.load_user_profile_screenshot",
lambda name: QPixmap(),
"bec_widgets.widgets.containers.dock_area.dock_area.load_runtime_profile_screenshot",
lambda name, namespace=None: QPixmap(),
)
monkeypatch.setattr(
"bec_widgets.widgets.containers.dock_area.dock_area.load_default_profile_screenshot",
lambda name: QPixmap(),
"bec_widgets.widgets.containers.dock_area.dock_area.load_baseline_profile_screenshot",
lambda name, namespace=None: QPixmap(),
)
monkeypatch.setattr(
"bec_widgets.widgets.containers.dock_area.dock_area.RestoreProfileDialog.confirm",
@@ -1500,24 +1453,49 @@ class TestAdvancedDockAreaRestoreAndDialogs:
)
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.restore_user_from_default"
"bec_widgets.widgets.containers.dock_area.dock_area.restore_runtime_from_baseline"
) as mock_restore:
advanced_dock_area.restore_user_profile_from_default()
advanced_dock_area.restore_baseline_profile(show_dialog=True)
mock_restore.assert_not_called()
def test_restore_user_profile_from_default_no_target(self, advanced_dock_area, monkeypatch):
def test_restore_runtime_profile_from_baseline_without_dialog(self, advanced_dock_area):
profile_name = "alignment_scan"
helper = profile_helper(advanced_dock_area)
helper.open_baseline(profile_name).sync()
helper.open_runtime(profile_name).sync()
with (
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.RestoreProfileDialog.confirm"
) as mock_confirm,
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.restore_runtime_from_baseline"
) as mock_restore,
patch.object(advanced_dock_area, "delete_all") as mock_delete_all,
patch.object(advanced_dock_area, "load_profile") as mock_load_profile,
):
advanced_dock_area.restore_baseline_profile(profile_name, show_dialog=False)
mock_confirm.assert_not_called()
mock_restore.assert_called_once_with(
profile_name, namespace=advanced_dock_area.profile_namespace
)
mock_delete_all.assert_called_once()
mock_load_profile.assert_called_once_with(profile_name)
def test_restore_runtime_profile_from_baseline_no_target(self, advanced_dock_area, monkeypatch):
advanced_dock_area._current_profile_name = None
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.RestoreProfileDialog.confirm"
) as mock_confirm:
advanced_dock_area.restore_user_profile_from_default()
advanced_dock_area.restore_baseline_profile(show_dialog=True)
mock_confirm.assert_not_called()
def test_refresh_workspace_list_with_refresh_profiles(self, advanced_dock_area):
profile_name = "refresh_profile"
helper = profile_helper(advanced_dock_area)
helper.open_user(profile_name).sync()
helper.open_runtime(profile_name).sync()
# Simulate a normal named-profile state (not transient empty startup mode).
advanced_dock_area._empty_profile_active = False
advanced_dock_area._current_profile_name = profile_name
@@ -1572,8 +1550,8 @@ class TestAdvancedDockAreaRestoreAndDialogs:
active = "active_profile"
quick = "quick_profile"
helper = profile_helper(advanced_dock_area)
helper.open_user(active).sync()
helper.open_user(quick).sync()
helper.open_runtime(active).sync()
helper.open_runtime(quick).sync()
helper.set_quick_select(quick, True)
combo_stub = ComboStub()
@@ -1600,7 +1578,7 @@ class TestAdvancedDockAreaRestoreAndDialogs:
advanced_dock_area._current_profile_name = "manager_profile"
helper = profile_helper(advanced_dock_area)
helper.open_user("manager_profile").sync()
helper.open_runtime("manager_profile").sync()
advanced_dock_area.show_workspace_manager()
@@ -1635,18 +1613,104 @@ class TestProfileManagement:
def test_profile_path(self, temp_profile_dir):
"""Test profile path generation."""
path = user_profile_path("test_profile")
expected = os.path.join(temp_profile_dir, "user", "test_profile.ini")
path = runtime_profile_path("test_profile")
expected = os.path.join(temp_profile_dir, "runtime", "test_profile.ini")
assert path == expected
default_path = default_profile_path("test_profile")
expected_default = os.path.join(temp_profile_dir, "default", "test_profile.ini")
assert default_path == expected_default
baseline_path = baseline_profile_path("test_profile")
expected_baseline = os.path.join(temp_profile_dir, "baseline", "test_profile.ini")
assert baseline_path == expected_baseline
def test_open_settings(self, temp_profile_dir):
"""Test opening settings for a profile."""
settings = open_user_settings("test_profile")
assert isinstance(settings, QSettings)
def test_legacy_user_profile_is_mapped_to_runtime(self, temp_profile_dir):
"""Legacy user profiles are copied into the canonical runtime segment."""
name = "legacy_runtime"
legacy_dir = os.path.join(temp_profile_dir, "user")
os.makedirs(legacy_dir, exist_ok=True)
legacy_path = os.path.join(legacy_dir, f"{name}.ini")
legacy_settings = QSettings(legacy_path, QSettings.IniFormat)
legacy_settings.setValue("test/value", "legacy")
legacy_settings.sync()
canonical_path = runtime_profile_path(name)
assert not os.path.exists(canonical_path)
assert name in list_profiles()
assert os.path.exists(canonical_path)
assert open_runtime_settings(name).value("test/value", "", type=str) == "legacy"
def test_legacy_default_profile_is_mapped_to_baseline(self, temp_profile_dir):
"""Legacy default profiles are copied into the canonical baseline segment."""
name = "legacy_baseline"
legacy_dir = os.path.join(temp_profile_dir, "default")
os.makedirs(legacy_dir, exist_ok=True)
legacy_path = os.path.join(legacy_dir, f"{name}.ini")
legacy_settings = QSettings(legacy_path, QSettings.IniFormat)
legacy_settings.setValue("test/value", "legacy")
legacy_settings.sync()
canonical_path = baseline_profile_path(name)
assert not os.path.exists(canonical_path)
assert name in list_profiles()
assert os.path.exists(canonical_path)
assert open_baseline_settings(name).value("test/value", "", type=str) == "legacy"
def test_runtime_namespace_fallback_is_materialized(self, temp_profile_dir):
"""Canonical runtime namespace fallback is copied before opening primary settings."""
name = "runtime_namespace_fallback"
fallback_settings = open_runtime_settings(name)
fallback_settings.setValue("test/value", "fallback")
fallback_settings.sync()
namespaced_path = runtime_profile_path(name, namespace="beamline")
assert not os.path.exists(namespaced_path)
settings = open_runtime_settings(name, namespace="beamline")
assert os.path.exists(namespaced_path)
assert settings.value("test/value", "", type=str) == "fallback"
def test_baseline_namespace_fallback_is_materialized(self, temp_profile_dir):
"""Canonical baseline namespace fallback is copied before opening primary settings."""
name = "baseline_namespace_fallback"
fallback_settings = open_baseline_settings(name)
fallback_settings.setValue("test/value", "fallback")
fallback_settings.sync()
namespaced_path = baseline_profile_path(name, namespace="beamline")
assert not os.path.exists(namespaced_path)
settings = open_baseline_settings(name, namespace="beamline")
assert os.path.exists(namespaced_path)
assert settings.value("test/value", "", type=str) == "fallback"
def test_canonical_profile_wins_over_legacy_profile(self, temp_profile_dir):
"""Canonical runtime/baseline files are not overwritten by legacy fallback files."""
name = "canonical_wins"
runtime_settings = open_runtime_settings(name)
runtime_settings.setValue("test/value", "canonical-runtime")
runtime_settings.sync()
baseline_settings = open_baseline_settings(name)
baseline_settings.setValue("test/value", "canonical-baseline")
baseline_settings.sync()
for segment, value in (("user", "legacy-runtime"), ("default", "legacy-baseline")):
legacy_dir = os.path.join(temp_profile_dir, segment)
os.makedirs(legacy_dir, exist_ok=True)
legacy_settings = QSettings(
os.path.join(legacy_dir, f"{name}.ini"), QSettings.IniFormat
)
legacy_settings.setValue("test/value", value)
legacy_settings.sync()
assert name in list_profiles()
assert open_runtime_settings(name).value("test/value", "", type=str) == "canonical-runtime"
assert (
open_baseline_settings(name).value("test/value", "", type=str) == "canonical-baseline"
)
def test_list_profiles_empty(self, temp_profile_dir):
"""Test listing profiles when directory is empty."""
@@ -1666,7 +1730,7 @@ class TestProfileManagement:
# Create some test profile files
profile_names = ["profile1", "profile2", "profile3"]
for name in profile_names:
settings = open_user_settings(name)
settings = open_runtime_settings(name)
settings.setValue("test", "value")
settings.sync()
@@ -1676,24 +1740,24 @@ class TestProfileManagement:
def test_readonly_profile_operations(self, temp_profile_dir, module_profile_factory):
"""Test read-only profile functionality."""
profile_name = "user_profile"
profile_name = "runtime_profile"
# Initially should not be read-only
assert not is_profile_read_only(profile_name)
# Create a user profile and ensure it's writable
settings = open_user_settings(profile_name)
# Create a runtime profile and ensure it's writable
settings = open_runtime_settings(profile_name)
settings.setValue("test", "value")
settings.sync()
assert not is_profile_read_only(profile_name)
# Verify a bundled module profile is detected as read-only
readonly_name = module_profile_factory("module_default")
readonly_name = module_profile_factory("module_baseline")
assert is_profile_read_only(readonly_name)
def test_write_and_read_manifest(self, temp_profile_dir, advanced_dock_area, qtbot):
"""Test writing and reading dock manifest."""
settings = open_user_settings("test_manifest")
settings = open_runtime_settings("test_manifest")
# Create real docks
advanced_dock_area.new("DarkModeButton")
@@ -1723,18 +1787,18 @@ class TestProfileManagement:
def test_restore_preserves_quick_select(self, temp_profile_dir):
"""Ensure restoring keeps the quick select flag when it was enabled."""
profile_name = "restorable_profile"
default_settings = open_default_settings(profile_name)
default_settings.setValue("test", "default")
default_settings.sync()
baseline_settings = open_baseline_settings(profile_name)
baseline_settings.setValue("test", "baseline")
baseline_settings.sync()
user_settings = open_user_settings(profile_name)
user_settings.setValue("test", "user")
user_settings.sync()
runtime_settings = open_runtime_settings(profile_name)
runtime_settings.setValue("test", "runtime")
runtime_settings.sync()
set_quick_select(profile_name, True)
assert is_quick_select(profile_name)
restore_user_from_default(profile_name)
restore_runtime_from_baseline(profile_name)
assert is_quick_select(profile_name)
@@ -1758,7 +1822,7 @@ class TestWorkspaceProfileOperations:
widget.prepare_for_shutdown()
mock_write.assert_not_called()
helper.open_user("real_profile").sync()
helper.open_runtime("real_profile").sync()
widget.load_profile("real_profile")
assert widget._empty_profile_active is False
assert widget._empty_profile_consumed is True
@@ -1772,7 +1836,7 @@ class TestWorkspaceProfileOperations:
profile_name = module_profile_factory("readonly_profile")
new_profile = f"{profile_name}_custom"
helper = profile_helper(advanced_dock_area)
target_path = helper.user_path(new_profile)
target_path = helper.runtime_path(new_profile)
if os.path.exists(target_path):
os.remove(target_path)
@@ -1802,7 +1866,7 @@ class TestWorkspaceProfileOperations:
helper = profile_helper(advanced_dock_area)
# Create a profile with manifest
settings = helper.open_user(profile_name)
settings = helper.open_runtime(profile_name)
settings.beginWriteArray("manifest/widgets", 1)
settings.setArrayIndex(0)
settings.setValue("object_name", "test_widget")
@@ -1823,6 +1887,83 @@ class TestWorkspaceProfileOperations:
widget_map = advanced_dock_area.widget_map()
assert "test_widget" in widget_map
def test_load_profile_default_does_not_restore_baseline(self, advanced_dock_area):
"""Regular profile loading should not restore the runtime copy."""
profile_name = "load_without_baseline_restore"
helper = profile_helper(advanced_dock_area)
helper.open_runtime(profile_name).sync()
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.restore_runtime_from_baseline"
) as mock_restore:
advanced_dock_area.load_profile(profile_name)
mock_restore.assert_not_called()
assert advanced_dock_area._current_profile_name == profile_name
def test_load_profile_restores_baseline_without_dialog(self, advanced_dock_area):
"""CLI loading can restore the runtime copy from baseline without confirmation."""
profile_name = "alignment_scan"
helper = profile_helper(advanced_dock_area)
helper.open_baseline(profile_name).sync()
helper.open_runtime(profile_name).sync()
with (
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.RestoreProfileDialog.confirm"
) as mock_confirm,
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.restore_runtime_from_baseline"
) as mock_restore,
):
advanced_dock_area.load_profile(profile_name, restore_baseline=True)
mock_confirm.assert_not_called()
mock_restore.assert_called_once_with(
profile_name, namespace=advanced_dock_area.profile_namespace
)
assert advanced_dock_area._current_profile_name == profile_name
def test_load_profile_materializes_runtime_namespace_fallback(self, advanced_dock_area):
"""Loading a runtime fallback copies it into the active namespace before opening."""
profile_name = "load_runtime_namespace_fallback"
helper = profile_helper(advanced_dock_area)
fallback_settings = open_runtime_settings(profile_name)
fallback_settings.setValue("test/value", "fallback")
fallback_settings.sync()
namespaced_path = helper.runtime_path(profile_name)
assert not os.path.exists(namespaced_path)
advanced_dock_area.load_profile(profile_name)
assert os.path.exists(namespaced_path)
assert (
QSettings(namespaced_path, QSettings.IniFormat).value("test/value", "", type=str)
== "fallback"
)
assert advanced_dock_area._current_profile_name == profile_name
def test_load_profile_materializes_baseline_namespace_fallback(self, advanced_dock_area):
"""Loading a baseline fallback copies it into the active namespace before opening."""
profile_name = "load_baseline_namespace_fallback"
helper = profile_helper(advanced_dock_area)
fallback_settings = open_baseline_settings(profile_name)
fallback_settings.setValue("test/value", "fallback")
fallback_settings.sync()
namespaced_path = helper.baseline_path(profile_name)
assert not os.path.exists(namespaced_path)
advanced_dock_area.load_profile(profile_name)
assert os.path.exists(namespaced_path)
assert (
QSettings(namespaced_path, QSettings.IniFormat).value("test/value", "", type=str)
== "fallback"
)
assert advanced_dock_area._current_profile_name == profile_name
def test_save_as_skips_autosave_source_profile(
self, advanced_dock_area, temp_profile_dir, qtbot
):
@@ -1831,7 +1972,7 @@ class TestWorkspaceProfileOperations:
new_profile = "autosave_new"
helper = profile_helper(advanced_dock_area)
settings = helper.open_user(source_profile)
settings = helper.open_runtime(source_profile)
settings.beginWriteArray("manifest/widgets", 1)
settings.setArrayIndex(0)
settings.setValue("object_name", "source_widget")
@@ -1863,11 +2004,16 @@ class TestWorkspaceProfileOperations:
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.SaveProfileDialog", StubDialog
):
advanced_dock_area.save_profile(show_dialog=True)
widgets_before_save = list(advanced_dock_area.widget_list())
with patch.object(advanced_dock_area, "load_profile") as mock_load_profile:
advanced_dock_area.save_profile(show_dialog=True)
qtbot.wait(100)
mock_load_profile.assert_not_called()
qtbot.wait(500)
source_manifest = read_manifest(helper.open_user(source_profile))
new_manifest = read_manifest(helper.open_user(new_profile))
assert list(advanced_dock_area.widget_list()) == widgets_before_save
source_manifest = read_manifest(helper.open_runtime(source_profile))
new_manifest = read_manifest(helper.open_runtime(new_profile))
assert len(source_manifest) == 1
assert len(new_manifest) == 2
@@ -1879,7 +2025,7 @@ class TestWorkspaceProfileOperations:
helper = profile_helper(advanced_dock_area)
for profile in (profile_a, profile_b):
settings = helper.open_user(profile)
settings = helper.open_runtime(profile)
settings.beginWriteArray("manifest/widgets", 1)
settings.setArrayIndex(0)
settings.setValue("object_name", f"{profile}_widget")
@@ -1898,7 +2044,7 @@ class TestWorkspaceProfileOperations:
advanced_dock_area.load_profile(profile_b)
qtbot.wait(500)
manifest_a = read_manifest(helper.open_user(profile_a))
manifest_a = read_manifest(helper.open_runtime(profile_a))
assert len(manifest_a) == 2
def test_delete_profile_readonly(
@@ -1907,15 +2053,15 @@ class TestWorkspaceProfileOperations:
"""Test deleting bundled profile removes only the writable copy."""
profile_name = module_profile_factory("readonly_profile")
helper = profile_helper(advanced_dock_area)
helper.list_profiles() # ensure default and user copies are materialized
helper.open_default(profile_name).sync()
settings = helper.open_user(profile_name)
helper.list_profiles() # ensure baseline and runtime copies are materialized
helper.open_baseline(profile_name).sync()
settings = helper.open_runtime(profile_name)
settings.setValue("test", "value")
settings.sync()
user_path = helper.user_path(profile_name)
default_path = helper.default_path(profile_name)
assert os.path.exists(user_path)
assert os.path.exists(default_path)
runtime_path = helper.runtime_path(profile_name)
baseline_path = helper.baseline_path(profile_name)
assert os.path.exists(runtime_path)
assert os.path.exists(baseline_path)
with patch.object(advanced_dock_area.toolbar.components, "get_action") as mock_get_action:
mock_combo = MagicMock()
@@ -1936,9 +2082,9 @@ class TestWorkspaceProfileOperations:
mock_question.assert_not_called()
mock_info.assert_called_once()
# Read-only profile should remain intact (user + default copies)
assert os.path.exists(user_path)
assert os.path.exists(default_path)
# Read-only profile should remain intact (runtime + baseline copies)
assert os.path.exists(runtime_path)
assert os.path.exists(baseline_path)
def test_delete_profile_success(self, advanced_dock_area, temp_profile_dir):
"""Test successful profile deletion."""
@@ -1946,11 +2092,11 @@ class TestWorkspaceProfileOperations:
helper = profile_helper(advanced_dock_area)
# Create regular profile
settings = helper.open_user(profile_name)
settings = helper.open_runtime(profile_name)
settings.setValue("test", "value")
settings.sync()
user_path = helper.user_path(profile_name)
assert os.path.exists(user_path)
runtime_path = helper.runtime_path(profile_name)
assert os.path.exists(runtime_path)
with patch.object(advanced_dock_area.toolbar.components, "get_action") as mock_get_action:
mock_combo = MagicMock()
@@ -1968,7 +2114,7 @@ class TestWorkspaceProfileOperations:
mock_question.assert_called_once()
mock_refresh.assert_called_once()
# Profile should be deleted
assert not os.path.exists(user_path)
assert not os.path.exists(runtime_path)
def test_delete_profile_cli_usage(self, advanced_dock_area, temp_profile_dir):
"""Test delete_profile with explicit name (CLI usage - no dialog by default)."""
@@ -1976,24 +2122,24 @@ class TestWorkspaceProfileOperations:
helper = profile_helper(advanced_dock_area)
# Create regular profile
settings = helper.open_user(profile_name)
settings = helper.open_runtime(profile_name)
settings.setValue("test", "value")
settings.sync()
user_path = helper.user_path(profile_name)
assert os.path.exists(user_path)
runtime_path = helper.runtime_path(profile_name)
assert os.path.exists(runtime_path)
# Delete without dialog (CLI usage - default behavior)
result = advanced_dock_area.delete_profile(profile_name)
assert result is True
assert not os.path.exists(user_path)
assert not os.path.exists(runtime_path)
def test_refresh_workspace_list(self, advanced_dock_area, temp_profile_dir):
"""Test refreshing workspace list."""
# Create some profiles
helper = profile_helper(advanced_dock_area)
for name in ["profile1", "profile2"]:
settings = helper.open_user(name)
settings = helper.open_runtime(name)
settings.setValue("test", "value")
settings.sync()
@@ -2033,20 +2179,6 @@ class TestCleanupAndMisc:
# Verify dock was removed
assert len(advanced_dock_area.dock_list()) == initial_count - 1
def test_apply_dock_lock(self, advanced_dock_area, qtbot):
"""Test _apply_dock_lock functionality."""
# Create a dock first
advanced_dock_area.new("DarkModeButton")
qtbot.wait(200)
# Test locking
advanced_dock_area._apply_dock_lock(True)
# No assertion needed - just verify it doesn't crash
# Test unlocking
advanced_dock_area._apply_dock_lock(False)
# No assertion needed - just verify it doesn't crash
def test_make_dock(self, advanced_dock_area):
"""Test _make_dock functionality."""
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import (
+1 -3
View File
@@ -247,12 +247,10 @@ def test_bec_weblinks(monkeypatch):
monkeypatch.setattr(webbrowser, "open", fake_open)
BECWebLinksMixin.open_bec_docs()
BECWebLinksMixin.open_bec_widgets_docs()
BECWebLinksMixin.open_bec_bug_report()
assert opened_urls == [
"https://beamline-experiment-control.readthedocs.io/en/latest/",
"https://bec.readthedocs.io/projects/bec-widgets/en/latest/",
"https://bec-project.github.io/bec_docs/",
"https://github.com/bec-project/bec_widgets/issues",
]
+2
View File
@@ -501,6 +501,7 @@ def test_changing_scans_remember_parameters(scan_control, mocked_client):
assert grid_kwargs["burst_at_each_point"] == kwargs["burst_at_each_point"]
@pytest.mark.skip(reason="Unreliable - GH issue #1134")
def test_get_scan_parameters_from_redis(scan_control, mocked_client):
scan_name = "line_scan"
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
@@ -585,6 +586,7 @@ def test_scan_metadata_is_passed_to_scan_function(scan_control: ScanControl):
scans.grid_scan.assert_called_once_with(metadata=TEST_MD)
@pytest.mark.skip(reason="Unreliable - GH issue #1134")
def test_restore_parameters_with_fewer_arg_bundles(scan_control, qtbot):
"""
Ensure that when more argument bundles are present than exist in the