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

Compare commits

..

52 Commits

Author SHA1 Message Date
wyzula_j 21ef51b5ab fix: test bw-generate-cli 2026-05-04 16:13:02 +02:00
wyzula_j b8a191c8be fix(dock_area): icon fetching for toolbar import optimised 2026-05-04 14:49:49 +02:00
wyzula_j 100a827922 fix(jupyter_console_widget): widget_handler API fix 2026-05-04 14:30:39 +02:00
wakonig_k bfe7ddfd74 feat: move to lazy widget import 2026-05-04 14:30:39 +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
semantic-release f13fa75e25 3.7.0
Automatically generated by python-semantic-release
2026-04-21 10:42:54 +00:00
wakonig_k 0cf84cd1d8 feat: move companion app to applications 2026-04-21 12:42:08 +02:00
wakonig_k 3e77f54034 refactor: cleanup of imports 2026-04-21 12:42:08 +02:00
semantic-release f7616102d8 3.6.0
Automatically generated by python-semantic-release
2026-04-21 06:39:15 +00:00
perl_d 5a497c3598 fix: small usability changes 2026-04-21 08:38:24 +02:00
perl_d 23e3644619 feat: add button/slot to pause/unpause logs 2026-04-21 08:38:24 +02:00
perl_d a5db2dc340 fix: change resize mode to interactive 2026-04-21 08:38:24 +02:00
perl_d 2e8f43fcac feat: add logpanel to menu 2026-04-21 08:38:24 +02:00
perl_d 09bb1121d8 feat: migrate logpanel to table model/view 2026-04-21 08:38:24 +02:00
semantic-release c9aaa77b3c 3.5.1
Automatically generated by python-semantic-release
2026-04-20 13:06:31 +00:00
perl_d f7a1ee49a4 fix: don't assume attr exists if we timed out waiting for it 2026-04-20 15:05:47 +02:00
perl_d 8e51c1adb6 refactor: don't import real widgets in client 2026-04-19 16:05:56 +02:00
semantic-release 846b6e6968 3.5.0
Automatically generated by python-semantic-release
2026-04-14 15:29:09 +00:00
perl_d f562c61e3c fix: connect signals the correct way around 2026-04-14 17:28:19 +02:00
wyzula_j bda5d38965 refactor: code cleanup 2026-04-14 17:28:19 +02:00
wyzula_j 9b0ec9dd79 fix(bec_console): persistent bec session 2026-04-14 17:28:19 +02:00
perl_d 1754e759f0 fix: create new bec shell if deleted 2026-04-14 17:28:19 +02:00
perl_d 308e84d0e1 tests: update tests 2026-04-14 17:28:19 +02:00
perl_d fa2ef83bb9 fix: formatting in plugin template 2026-04-14 17:28:19 +02:00
perl_d 02cb393bb0 feat: add qtermwidget plugin and replace web term 2026-04-14 17:28:19 +02:00
semantic-release 1d3e0214fd 3.4.4
Automatically generated by python-semantic-release
2026-04-14 07:33:15 +00:00
perl_d 37747babda fix: check for duplicate subscriptions in GUIClient 2026-04-14 09:32:17 +02:00
perl_d 32f5d486d3 fix: make gui client registry callback non static 2026-04-14 09:32:17 +02:00
perl_d 0ff1fdc815 fix: remove staticmethod subscription 2026-04-14 09:32:17 +02:00
perl_d c7de320ca5 fix: check duplicate stream sub 2026-04-14 09:32:17 +02:00
semantic-release 5b23dce3d0 3.4.3
Automatically generated by python-semantic-release
2026-04-13 09:20:13 +00:00
wakonig_k 5e84d3bec6 fix: Set OPHYD_CONTROL_LAYER to dummy for tests 2026-04-13 11:19:22 +02:00
105 changed files with 3803 additions and 1638 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
+189
View File
@@ -1,6 +1,195 @@
# 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
- Move companion app to applications
([`0cf84cd`](https://github.com/bec-project/bec_widgets/commit/0cf84cd1d839ac4a39ffb5fb9ba57d432e04348a))
### Refactoring
- Cleanup of imports
([`3e77f54`](https://github.com/bec-project/bec_widgets/commit/3e77f540345f56b9f184a332fcdd50d4d4c8c621))
## v3.6.0 (2026-04-21)
### Bug Fixes
- Change resize mode to interactive
([`a5db2dc`](https://github.com/bec-project/bec_widgets/commit/a5db2dc340f3386e68b300fd4528a44f87cbbf97))
- Small usability changes
([`5a497c3`](https://github.com/bec-project/bec_widgets/commit/5a497c3598c2d8f27916d91d53c646d5d6d3a4a7))
### Features
- Add button/slot to pause/unpause logs
([`23e3644`](https://github.com/bec-project/bec_widgets/commit/23e3644619de958bcfdb8a0b2ee1f7c2ce05b235))
- Add logpanel to menu
([`2e8f43f`](https://github.com/bec-project/bec_widgets/commit/2e8f43fcac581cd1c227308198565d142a1bf276))
- Migrate logpanel to table model/view
([`09bb112`](https://github.com/bec-project/bec_widgets/commit/09bb1121d83bac1f6e4827daa476fbe7cd5b3a80))
## v3.5.1 (2026-04-20)
### Bug Fixes
- Don't assume attr exists if we timed out waiting for it
([`f7a1ee4`](https://github.com/bec-project/bec_widgets/commit/f7a1ee49a42c58ba315c8957b45a80d862ffe745))
### Refactoring
- Don't import real widgets in client
([`8e51c1a`](https://github.com/bec-project/bec_widgets/commit/8e51c1adb6a7658c54846794cf97b774cbac2193))
## v3.5.0 (2026-04-14)
### Bug Fixes
- Connect signals the correct way around
([`f562c61`](https://github.com/bec-project/bec_widgets/commit/f562c61e3cec3387f6821bad74403beeb3436355))
- Create new bec shell if deleted
([`1754e75`](https://github.com/bec-project/bec_widgets/commit/1754e759f0c59f2f4063f661bacd334127326947))
- Formatting in plugin template
([`fa2ef83`](https://github.com/bec-project/bec_widgets/commit/fa2ef83bb9dfeeb4c5fc7cd77168c16101c32693))
- **bec_console**: Persistent bec session
([`9b0ec9d`](https://github.com/bec-project/bec_widgets/commit/9b0ec9dd79ad1adc5d211dd703db7441da965f34))
### Features
- Add qtermwidget plugin and replace web term
([`02cb393`](https://github.com/bec-project/bec_widgets/commit/02cb393bb086165dc64917b633d5570d02e1a2a9))
### Refactoring
- Code cleanup
([`bda5d38`](https://github.com/bec-project/bec_widgets/commit/bda5d389651bb2b13734cd31159679e85b1bd583))
## v3.4.4 (2026-04-14)
### Bug Fixes
- Check duplicate stream sub
([`c7de320`](https://github.com/bec-project/bec_widgets/commit/c7de320ca564264a31b84931f553170f25659685))
- Check for duplicate subscriptions in GUIClient
([`37747ba`](https://github.com/bec-project/bec_widgets/commit/37747babda407040333c6bd04646be9a49e0ee81))
- Make gui client registry callback non static
([`32f5d48`](https://github.com/bec-project/bec_widgets/commit/32f5d486d3fc8d41df2668c58932ae982819b285))
- Remove staticmethod subscription
([`0ff1fdc`](https://github.com/bec-project/bec_widgets/commit/0ff1fdc81578eec3ffc5d4030fca7b357a0b4c2f))
## v3.4.3 (2026-04-13)
### Bug Fixes
- Set OPHYD_CONTROL_LAYER to dummy for tests
([`5e84d3b`](https://github.com/bec-project/bec_widgets/commit/5e84d3bec608ae9f2ee6dae67db2e3e1387b1f59))
## v3.4.2 (2026-04-01)
### Bug Fixes
+12 -18
View File
@@ -1,19 +1,13 @@
import os
import sys
import bec_widgets.widgets.containers.qt_ads as QtAds
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
if sys.platform.startswith("linux"):
qt_platform = os.environ.get("QT_QPA_PLATFORM", "")
if qt_platform != "offscreen":
os.environ["QT_QPA_PLATFORM"] = "xcb"
# Default QtAds configuration
QtAds.CDockManager.setConfigFlag(QtAds.CDockManager.eConfigFlag.FocusHighlighting, True)
QtAds.CDockManager.setConfigFlag(
QtAds.CDockManager.eConfigFlag.RetainTabSizeWhenCloseButtonHidden, True
)
__all__ = ["BECWidget", "SafeSlot", "SafeProperty"]
def __getattr__(name):
if name == "BECWidget":
from bec_widgets.utils.bec_widget import BECWidget
return BECWidget
if name in {"SafeSlot", "SafeProperty"}:
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
return {"SafeSlot": SafeSlot, "SafeProperty": SafeProperty}[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+15
View File
@@ -0,0 +1,15 @@
import os
import sys
import bec_widgets.widgets.containers.qt_ads as QtAds
if sys.platform.startswith("linux"):
qt_platform = os.environ.get("QT_QPA_PLATFORM", "")
if qt_platform != "offscreen":
os.environ["QT_QPA_PLATFORM"] = "xcb"
# Default QtAds configuration
QtAds.CDockManager.setConfigFlag(QtAds.CDockManager.eConfigFlag.FocusHighlighting, True)
QtAds.CDockManager.setConfigFlag(
QtAds.CDockManager.eConfigFlag.RetainTabSizeWhenCloseButtonHidden, True
)
@@ -19,8 +19,8 @@ from qtpy.QtWidgets import QApplication
import bec_widgets
from bec_widgets.applications.launch_window import LaunchWindow
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.rpc_register import RPCRegister
logger = bec_logger.logger
+1 -1
View File
@@ -20,13 +20,13 @@ from qtpy.QtWidgets import (
)
import bec_widgets
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.name_utils import pascal_to_snake
from bec_widgets.utils.plugin_utils import get_plugin_auto_updates
from bec_widgets.utils.round_frame import RoundedFrame
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.screen_utils import apply_window_geometry, centered_geometry_for_app
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.utils.ui_loader import UILoader
+1
View File
@@ -0,0 +1 @@
from bec_widgets.cli.rpc import rpc_base
+173 -42
View File
@@ -13,7 +13,7 @@ from typing import Literal, Optional
from bec_lib.logger import bec_logger
from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call, rpc_timeout
from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets, get_plugin_client_module
from bec_widgets.utils.bec_plugin_helper import get_plugin_client_module
logger = bec_logger.logger
@@ -62,29 +62,19 @@ _Widgets = {
try:
_plugin_widgets = get_all_plugin_widgets().as_dict()
plugin_client = get_plugin_client_module()
Widgets = _WidgetsEnumType("Widgets", {name: name for name in _plugin_widgets} | _Widgets)
if (_overlap := _Widgets.keys() & _plugin_widgets.keys()) != set():
for _widget in _overlap:
logger.warning(
f"Detected duplicate widget {_widget} in plugin repo file: {inspect.getfile(_plugin_widgets[_widget])} !"
)
for plugin_name, plugin_class in inspect.getmembers(plugin_client, inspect.isclass):
if issubclass(plugin_class, RPCBase) and plugin_class is not RPCBase:
if plugin_name not in _Widgets:
_Widgets[plugin_name] = plugin_name
if plugin_name in globals():
conflicting_file = (
inspect.getfile(_plugin_widgets[plugin_name])
if plugin_name in _plugin_widgets
else f"{plugin_client}"
)
logger.warning(
f"Plugin widget {plugin_name} from {conflicting_file} conflicts with a built-in class!"
f"Plugin widget {plugin_name} in {plugin_class._IMPORT_MODULE} conflicts with a built-in class!"
)
continue
if plugin_name not in _overlap:
else:
globals()[plugin_name] = plugin_class
Widgets = _WidgetsEnumType("Widgets", _Widgets)
except ImportError as e:
logger.error(f"Failed loading plugins: \n{reduce(add, traceback.format_exception(e))}")
@@ -92,6 +82,8 @@ except ImportError as e:
class AdminView(RPCBase):
"""A view for administrators to change the current active experiment, manage messaging"""
_IMPORT_MODULE = "bec_widgets.applications.views.admin_view.admin_view"
@rpc_call
def activate(self) -> "None":
"""
@@ -100,6 +92,8 @@ class AdminView(RPCBase):
class AutoUpdates(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.containers.auto_update.auto_updates"
@property
@rpc_call
def enabled(self) -> "bool":
@@ -136,6 +130,8 @@ class AutoUpdates(RPCBase):
class AvailableDeviceResources(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources"
@rpc_call
def remove(self):
"""
@@ -156,6 +152,8 @@ class AvailableDeviceResources(RPCBase):
class BECDockArea(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.containers.dock_area.dock_area"
@rpc_call
def new(
self,
@@ -342,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:
@@ -360,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
@@ -391,6 +405,8 @@ class BECDockArea(RPCBase):
class BECMainWindow(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.containers.main_window.main_window"
@rpc_call
def remove(self):
"""
@@ -413,6 +429,8 @@ class BECMainWindow(RPCBase):
class BECProgressBar(RPCBase):
"""A custom progress bar with smooth transitions. The displayed text can be customized using a template."""
_IMPORT_MODULE = "bec_widgets.widgets.progress.bec_progressbar.bec_progressbar"
@rpc_call
def set_value(self, value):
"""
@@ -486,6 +504,8 @@ class BECProgressBar(RPCBase):
class BECQueue(RPCBase):
"""Widget to display the BEC queue."""
_IMPORT_MODULE = "bec_widgets.widgets.services.bec_queue.bec_queue"
@rpc_call
def remove(self):
"""
@@ -508,6 +528,8 @@ class BECQueue(RPCBase):
class BECShell(RPCBase):
"""A BecConsole pre-configured to run the BEC shell."""
_IMPORT_MODULE = "bec_widgets.widgets.editors.bec_console.bec_console"
@rpc_call
def remove(self):
"""
@@ -530,6 +552,8 @@ class BECShell(RPCBase):
class BECStatusBox(RPCBase):
"""An autonomous widget to display the status of BEC services."""
_IMPORT_MODULE = "bec_widgets.widgets.services.bec_status_box.bec_status_box"
@rpc_call
def get_server_state(self) -> "str":
"""
@@ -565,6 +589,8 @@ class BECStatusBox(RPCBase):
class BaseROI(RPCBase):
"""Base class for all Region of Interest (ROI) implementations."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.roi.image_roi"
@property
@rpc_call
def label(self) -> "str":
@@ -694,6 +720,8 @@ class BaseROI(RPCBase):
class BecConsole(RPCBase):
"""A console widget with access to a shared registry of terminals, such that instances can be moved around."""
_IMPORT_MODULE = "bec_widgets.widgets.editors.bec_console.bec_console"
@rpc_call
def remove(self):
"""
@@ -716,6 +744,8 @@ class BecConsole(RPCBase):
class CircularROI(RPCBase):
"""Circular Region of Interest with center/diameter tracking and auto-labeling."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.roi.image_roi"
@property
@rpc_call
def label(self) -> "str":
@@ -843,6 +873,8 @@ class CircularROI(RPCBase):
class Curve(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.plots.waveform.curve"
@rpc_call
def remove(self):
"""
@@ -1009,6 +1041,8 @@ class Curve(RPCBase):
class DapComboBox(RPCBase):
"""Editable combobox listing the available DAP models."""
_IMPORT_MODULE = "bec_widgets.widgets.dap.dap_combo_box.dap_combo_box"
@rpc_call
def select_y_axis(self, y_axis: str):
"""
@@ -1040,6 +1074,8 @@ class DapComboBox(RPCBase):
class DeveloperView(RPCBase):
"""A view for users to write scripts and macros and execute them within the application."""
_IMPORT_MODULE = "bec_widgets.applications.views.developer_view.developer_view"
@rpc_call
def activate(self) -> "None":
"""
@@ -1050,6 +1086,8 @@ class DeveloperView(RPCBase):
class DeviceBrowser(RPCBase):
"""DeviceBrowser is a widget that displays all available devices in the current BEC session."""
_IMPORT_MODULE = "bec_widgets.widgets.services.device_browser.device_browser"
@rpc_call
def remove(self):
"""
@@ -1072,6 +1110,8 @@ class DeviceBrowser(RPCBase):
class DeviceInitializationProgressBar(RPCBase):
"""A progress bar that displays the progress of device initialization."""
_IMPORT_MODULE = "bec_widgets.widgets.progress.device_initialization_progress_bar.device_initialization_progress_bar"
@rpc_call
def remove(self):
"""
@@ -1094,6 +1134,8 @@ class DeviceInitializationProgressBar(RPCBase):
class DeviceInputBase(RPCBase):
"""Mixin base class for device input widgets."""
_IMPORT_MODULE = "bec_widgets.widgets.control.device_input.base_classes.device_input_base"
@rpc_call
def remove(self):
"""
@@ -1116,6 +1158,8 @@ class DeviceInputBase(RPCBase):
class DeviceManagerView(RPCBase):
"""A view for users to manage devices within the application."""
_IMPORT_MODULE = "bec_widgets.applications.views.device_manager_view.device_manager_view"
@rpc_call
def activate(self) -> "None":
"""
@@ -1126,6 +1170,8 @@ class DeviceManagerView(RPCBase):
class DockAreaView(RPCBase):
"""Modular dock area view for arranging and managing multiple dockable widgets."""
_IMPORT_MODULE = "bec_widgets.applications.views.dock_area_view.dock_area_view"
@rpc_call
def activate(self) -> "None":
"""
@@ -1318,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:
@@ -1336,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
@@ -1369,6 +1431,8 @@ class DockAreaView(RPCBase):
class DockAreaWidget(RPCBase):
"""Lightweight dock area that exposes the core Qt ADS docking helpers without any"""
_IMPORT_MODULE = "bec_widgets.widgets.containers.dock_area.basic_dock_area"
@rpc_call
def new(
self,
@@ -1553,6 +1617,8 @@ class DockAreaWidget(RPCBase):
class EllipticalROI(RPCBase):
"""Elliptical Region of Interest with centre/width/height tracking and auto-labelling."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.roi.image_roi"
@property
@rpc_call
def label(self) -> "str":
@@ -1675,6 +1741,8 @@ class EllipticalROI(RPCBase):
class Heatmap(RPCBase):
"""Heatmap widget for visualizing 2d grid data with color mapping for the z-axis."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.heatmap.heatmap"
@rpc_call
def remove(self):
"""
@@ -2373,6 +2441,8 @@ class Heatmap(RPCBase):
class Image(RPCBase):
"""Image widget for displaying 2D data."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.image.image"
@rpc_call
def remove(self):
"""
@@ -2984,6 +3054,8 @@ class Image(RPCBase):
class ImageItem(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.plots.image.image_item"
@property
@rpc_call
def color_map(self) -> "str":
@@ -3134,6 +3206,8 @@ class ImageItem(RPCBase):
class LaunchWindow(RPCBase):
_IMPORT_MODULE = "bec_widgets.applications.launch_window"
@rpc_call
def show_launcher(self):
"""
@@ -3148,33 +3222,38 @@ class LaunchWindow(RPCBase):
class LogPanel(RPCBase):
"""Displays a log panel"""
"""Live display of the BEC logs in a table view."""
_IMPORT_MODULE = "bec_widgets.widgets.utility.logpanel.logpanel"
@rpc_call
def set_plain_text(self, text: str) -> None:
def remove(self):
"""
Set the plain text of the widget.
Args:
text (str): The text to set.
Cleanup the BECConnector
"""
@rpc_call
def set_html_text(self, text: str) -> None:
def attach(self):
"""
None
"""
Set the HTML text of the widget.
Args:
text (str): The text to set.
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class Minesweeper(RPCBase): ...
class Minesweeper(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.games.minesweeper"
class MonacoDock(RPCBase):
"""MonacoDock is a dock widget that contains Monaco editor instances."""
_IMPORT_MODULE = "bec_widgets.widgets.editors.monaco.monaco_dock"
@rpc_call
def new(
self,
@@ -3359,6 +3438,8 @@ class MonacoDock(RPCBase):
class MonacoWidget(RPCBase):
"""A simple Monaco editor widget"""
_IMPORT_MODULE = "bec_widgets.widgets.editors.monaco.monaco_widget"
@rpc_call
def set_text(
self, text: "str", file_name: "str | None" = None, reset: "bool" = False
@@ -3533,6 +3614,8 @@ class MonacoWidget(RPCBase):
class MotorMap(RPCBase):
"""Motor map widget for plotting motor positions in 2D including a trace of the last points."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.motor_map.motor_map"
@rpc_call
def remove(self):
"""
@@ -4003,6 +4086,8 @@ class MotorMap(RPCBase):
class MultiWaveform(RPCBase):
"""MultiWaveform widget for displaying multiple waveforms emitted by a single signal."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.multi_waveform.multi_waveform"
@rpc_call
def remove(self):
"""
@@ -4462,6 +4547,8 @@ class MultiWaveform(RPCBase):
class PdfViewerWidget(RPCBase):
"""A widget to display PDF documents with toolbar controls."""
_IMPORT_MODULE = "bec_widgets.widgets.utility.pdf_viewer.pdf_viewer"
@rpc_call
def load_pdf(self, file_path: str):
"""
@@ -4593,6 +4680,10 @@ class PdfViewerWidget(RPCBase):
class PositionIndicator(RPCBase):
"""Display a position within a defined range, e.g. motor limits."""
_IMPORT_MODULE = (
"bec_widgets.widgets.control.device_control.position_indicator.position_indicator"
)
@rpc_call
def set_value(self, position: float):
"""
@@ -4658,6 +4749,10 @@ class PositionIndicator(RPCBase):
class PositionerBox(RPCBase):
"""Simple Widget to control a positioner in box form"""
_IMPORT_MODULE = (
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box.positioner_box"
)
@rpc_call
def set_positioner(self, positioner: "str | Positioner"):
"""
@@ -4690,6 +4785,8 @@ class PositionerBox(RPCBase):
class PositionerBox2D(RPCBase):
"""Simple Widget to control two positioners in box form"""
_IMPORT_MODULE = "bec_widgets.widgets.control.device_control.positioner_box.positioner_box_2d.positioner_box_2d"
@rpc_call
def set_positioner_hor(self, positioner: "str | Positioner"):
"""
@@ -4759,6 +4856,8 @@ class PositionerBox2D(RPCBase):
class PositionerControlLine(RPCBase):
"""A widget that controls a single device."""
_IMPORT_MODULE = "bec_widgets.widgets.control.device_control.positioner_box.positioner_control_line.positioner_control_line"
@rpc_call
def set_positioner(self, positioner: "str | Positioner"):
"""
@@ -4791,6 +4890,8 @@ class PositionerControlLine(RPCBase):
class PositionerGroup(RPCBase):
"""Simple Widget to control a positioner in box form"""
_IMPORT_MODULE = "bec_widgets.widgets.control.device_control.positioner_group.positioner_group"
@rpc_call
def set_positioners(self, device_names: "str"):
"""
@@ -4822,6 +4923,8 @@ class PositionerGroup(RPCBase):
class RectangularROI(RPCBase):
"""Defines a rectangular Region of Interest (ROI) with additional functionality."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.roi.image_roi"
@property
@rpc_call
def label(self) -> "str":
@@ -4951,6 +5054,8 @@ class RectangularROI(RPCBase):
class ResumeButton(RPCBase):
"""A button that continue scan queue."""
_IMPORT_MODULE = "bec_widgets.widgets.control.buttons.button_resume.button_resume"
@rpc_call
def remove(self):
"""
@@ -4971,6 +5076,8 @@ class ResumeButton(RPCBase):
class Ring(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.progress.ring_progress_bar.ring"
@rpc_call
def set_value(self, value: "int | float"):
"""
@@ -5064,6 +5171,8 @@ class Ring(RPCBase):
class RingProgressBar(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar"
@rpc_call
def remove(self):
"""
@@ -5143,12 +5252,14 @@ class RingProgressBar(RPCBase):
class SBBMonitor(RPCBase):
"""A widget to display the SBB monitor website."""
...
_IMPORT_MODULE = "bec_widgets.widgets.editors.sbb_monitor.sbb_monitor"
class ScanControl(RPCBase):
"""Widget to submit new scans to the queue."""
_IMPORT_MODULE = "bec_widgets.widgets.control.scan_control.scan_control"
@rpc_call
def attach(self):
"""
@@ -5172,6 +5283,8 @@ class ScanControl(RPCBase):
class ScanProgressBar(RPCBase):
"""Widget to display a progress bar that is hooked up to the scan progress of a scan."""
_IMPORT_MODULE = "bec_widgets.widgets.progress.scan_progressbar.scan_progressbar"
@rpc_call
def remove(self):
"""
@@ -5194,6 +5307,8 @@ class ScanProgressBar(RPCBase):
class ScatterCurve(RPCBase):
"""Scatter curve item for the scatter waveform widget."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.scatter_waveform.scatter_curve"
@property
@rpc_call
def color_map(self) -> "str":
@@ -5203,6 +5318,8 @@ class ScatterCurve(RPCBase):
class ScatterWaveform(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.plots.scatter_waveform.scatter_waveform"
@rpc_call
def remove(self):
"""
@@ -5670,6 +5787,8 @@ class ScatterWaveform(RPCBase):
class SignalLabel(RPCBase):
_IMPORT_MODULE = "bec_widgets.widgets.utility.signal_label.signal_label"
@property
@rpc_call
def custom_label(self) -> "str":
@@ -5814,6 +5933,8 @@ class SignalLabel(RPCBase):
class TextBox(RPCBase):
"""A widget that displays text in plain and HTML format"""
_IMPORT_MODULE = "bec_widgets.widgets.editors.text_box.text_box"
@rpc_call
def set_plain_text(self, text: str) -> None:
"""
@@ -5836,6 +5957,8 @@ class TextBox(RPCBase):
class ViewBase(RPCBase):
"""Wrapper for a content widget used inside the main app's stacked view."""
_IMPORT_MODULE = "bec_widgets.applications.views.view"
@rpc_call
def activate(self) -> "None":
"""
@@ -5846,6 +5969,8 @@ class ViewBase(RPCBase):
class Waveform(RPCBase):
"""Widget for plotting waveforms."""
_IMPORT_MODULE = "bec_widgets.widgets.plots.waveform.waveform"
@rpc_call
def remove(self):
"""
@@ -6424,6 +6549,8 @@ class Waveform(RPCBase):
class WaveformViewInline(RPCBase):
_IMPORT_MODULE = "bec_widgets.applications.views.view"
@rpc_call
def activate(self) -> "None":
"""
@@ -6432,6 +6559,8 @@ class WaveformViewInline(RPCBase):
class WaveformViewPopup(RPCBase):
_IMPORT_MODULE = "bec_widgets.applications.views.view"
@rpc_call
def activate(self) -> "None":
"""
@@ -6442,6 +6571,8 @@ class WaveformViewPopup(RPCBase):
class WebsiteWidget(RPCBase):
"""A simple widget to display a website"""
_IMPORT_MODULE = "bec_widgets.widgets.editors.website.website"
@rpc_call
def set_url(self, url: str) -> None:
"""
+11 -13
View File
@@ -10,9 +10,9 @@ import threading
import time
from contextlib import contextmanager
from threading import Lock
from typing import TYPE_CHECKING, Literal, TypeAlias, cast
from typing import TYPE_CHECKING, Callable, Literal, TypeAlias, cast
from bec_lib.endpoints import MessageEndpoints
from bec_lib.endpoints import EndpointInfo, MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from rich.console import Console
@@ -232,6 +232,11 @@ class BECGuiClient(RPCBase):
"""The launcher object."""
return RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self, object_name="launcher")
def _safe_register_stream(self, endpoint: EndpointInfo, cb: Callable, **kwargs):
"""Check if already registered for registration in idempotent functions."""
if not self._client.connector.any_stream_is_registered(endpoint, cb=cb):
self._client.connector.register(endpoint, cb=cb, **kwargs)
def connect_to_gui_server(self, gui_id: str) -> None:
"""Connect to a GUI server"""
# Unregister the old callback
@@ -247,10 +252,9 @@ class BECGuiClient(RPCBase):
self._ipython_registry = {}
# Register the new callback
self._client.connector.register(
self._safe_register_stream(
MessageEndpoints.gui_registry_state(self._gui_id),
cb=self._handle_registry_update,
parent=self,
from_start=True,
)
@@ -531,20 +535,14 @@ class BECGuiClient(RPCBase):
def _start(self, wait: bool = False) -> None:
self._killed = False
self._client.connector.register(
MessageEndpoints.gui_registry_state(self._gui_id),
cb=self._handle_registry_update,
parent=self,
self._safe_register_stream(
MessageEndpoints.gui_registry_state(self._gui_id), cb=self._handle_registry_update
)
return self._start_server(wait=wait)
@staticmethod
def _handle_registry_update(
msg: dict[str, GUIRegistryStateMessage], parent: BECGuiClient
) -> None:
def _handle_registry_update(self, msg: dict[str, GUIRegistryStateMessage]) -> None:
# This was causing a deadlock during shutdown, not sure why.
# with self._lock:
self = parent
self._server_registry = cast(dict[str, RegistryState], msg["data"].state)
self._update_dynamic_namespace(self._server_registry)
+171
View File
@@ -0,0 +1,171 @@
# This file was automatically generated by generate_cli.py
# type: ignore
from __future__ import annotations
# pylint: skip-file
designer_plugins = {
"AbortButton": ("bec_widgets.widgets.control.buttons.button_abort.button_abort", "AbortButton"),
"BECColorMapWidget": (
"bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget",
"BECColorMapWidget",
),
"BECMainWindow": ("bec_widgets.widgets.containers.main_window.main_window", "BECMainWindow"),
"BECProgressBar": (
"bec_widgets.widgets.progress.bec_progressbar.bec_progressbar",
"BECProgressBar",
),
"BECQueue": ("bec_widgets.widgets.services.bec_queue.bec_queue", "BECQueue"),
"BECShell": ("bec_widgets.widgets.editors.bec_console.bec_console", "BECShell"),
"BECSpinBox": ("bec_widgets.widgets.utility.spinbox.decimal_spinbox", "BECSpinBox"),
"BECStatusBox": ("bec_widgets.widgets.services.bec_status_box.bec_status_box", "BECStatusBox"),
"BecConsole": ("bec_widgets.widgets.editors.bec_console.bec_console", "BecConsole"),
"ColorButton": ("bec_widgets.widgets.utility.visual.color_button.color_button", "ColorButton"),
"ColorButtonNative": (
"bec_widgets.widgets.utility.visual.color_button_native.color_button_native",
"ColorButtonNative",
),
"ColormapSelector": (
"bec_widgets.widgets.utility.visual.colormap_selector.colormap_selector",
"ColormapSelector",
),
"DapComboBox": ("bec_widgets.widgets.dap.dap_combo_box.dap_combo_box", "DapComboBox"),
"DarkModeButton": (
"bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button",
"DarkModeButton",
),
"DeviceBrowser": (
"bec_widgets.widgets.services.device_browser.device_browser",
"DeviceBrowser",
),
"DeviceComboBox": (
"bec_widgets.widgets.control.device_input.device_combobox.device_combobox",
"DeviceComboBox",
),
"DeviceLineEdit": (
"bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit",
"DeviceLineEdit",
),
"Heatmap": ("bec_widgets.widgets.plots.heatmap.heatmap", "Heatmap"),
"IDEExplorer": ("bec_widgets.widgets.utility.ide_explorer.ide_explorer", "IDEExplorer"),
"Image": ("bec_widgets.widgets.plots.image.image", "Image"),
"LMFitDialog": ("bec_widgets.widgets.dap.lmfit_dialog.lmfit_dialog", "LMFitDialog"),
"LogPanel": ("bec_widgets.widgets.utility.logpanel.logpanel", "LogPanel"),
"Minesweeper": ("bec_widgets.widgets.games.minesweeper", "Minesweeper"),
"MonacoWidget": ("bec_widgets.widgets.editors.monaco.monaco_widget", "MonacoWidget"),
"MotorMap": ("bec_widgets.widgets.plots.motor_map.motor_map", "MotorMap"),
"MultiWaveform": ("bec_widgets.widgets.plots.multi_waveform.multi_waveform", "MultiWaveform"),
"PdfViewerWidget": ("bec_widgets.widgets.utility.pdf_viewer.pdf_viewer", "PdfViewerWidget"),
"PositionIndicator": (
"bec_widgets.widgets.control.device_control.position_indicator.position_indicator",
"PositionIndicator",
),
"PositionerBox": (
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box.positioner_box",
"PositionerBox",
),
"PositionerBox2D": (
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_2d.positioner_box_2d",
"PositionerBox2D",
),
"PositionerControlLine": (
"bec_widgets.widgets.control.device_control.positioner_box.positioner_control_line.positioner_control_line",
"PositionerControlLine",
),
"PositionerGroup": (
"bec_widgets.widgets.control.device_control.positioner_group.positioner_group",
"PositionerGroup",
),
"ResetButton": ("bec_widgets.widgets.control.buttons.button_reset.button_reset", "ResetButton"),
"ResumeButton": (
"bec_widgets.widgets.control.buttons.button_resume.button_resume",
"ResumeButton",
),
"RingProgressBar": (
"bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar",
"RingProgressBar",
),
"SBBMonitor": ("bec_widgets.widgets.editors.sbb_monitor.sbb_monitor", "SBBMonitor"),
"ScanControl": ("bec_widgets.widgets.control.scan_control.scan_control", "ScanControl"),
"ScanMetadata": ("bec_widgets.widgets.editors.scan_metadata.scan_metadata", "ScanMetadata"),
"ScanProgressBar": (
"bec_widgets.widgets.progress.scan_progressbar.scan_progressbar",
"ScanProgressBar",
),
"ScatterWaveform": (
"bec_widgets.widgets.plots.scatter_waveform.scatter_waveform",
"ScatterWaveform",
),
"SignalComboBox": (
"bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox",
"SignalComboBox",
),
"SignalLabel": ("bec_widgets.widgets.utility.signal_label.signal_label", "SignalLabel"),
"SignalLineEdit": (
"bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit",
"SignalLineEdit",
),
"SpinnerWidget": ("bec_widgets.widgets.utility.spinner.spinner", "SpinnerWidget"),
"StopButton": ("bec_widgets.widgets.control.buttons.stop_button.stop_button", "StopButton"),
"TextBox": ("bec_widgets.widgets.editors.text_box.text_box", "TextBox"),
"ToggleSwitch": ("bec_widgets.widgets.utility.toggle.toggle", "ToggleSwitch"),
"Waveform": ("bec_widgets.widgets.plots.waveform.waveform", "Waveform"),
"WebsiteWidget": ("bec_widgets.widgets.editors.website.website", "WebsiteWidget"),
"WidgetFinderComboBox": (
"bec_widgets.widgets.utility.widget_finder.widget_finder",
"WidgetFinderComboBox",
),
}
widget_icons = {
"AbortButton": "cancel",
"BECColorMapWidget": "palette",
"BECMainWindow": "widgets",
"BECProgressBar": "page_control",
"BECQueue": "edit_note",
"BECShell": "hub",
"BECSpinBox": "123",
"BECStatusBox": "widgets",
"BecConsole": "terminal",
"ColorButton": "colors",
"ColorButtonNative": "colors",
"ColormapSelector": "palette",
"DapComboBox": "data_exploration",
"DarkModeButton": "dark_mode",
"DeviceBrowser": "lists",
"DeviceComboBox": "list_alt",
"DeviceLineEdit": "edit_note",
"Heatmap": "dataset",
"IDEExplorer": "widgets",
"Image": "image",
"LMFitDialog": "monitoring",
"LogPanel": "browse_activity",
"Minesweeper": "videogame_asset",
"MonacoWidget": "code",
"MotorMap": "my_location",
"MultiWaveform": "ssid_chart",
"PdfViewerWidget": "picture_as_pdf",
"PositionIndicator": "horizontal_distribute",
"PositionerBox": "switch_right",
"PositionerBox2D": "switch_right",
"PositionerControlLine": "switch_left",
"PositionerGroup": "grid_view",
"ResetButton": "restart_alt",
"ResumeButton": "resume",
"RingProgressBar": "track_changes",
"SBBMonitor": "train",
"ScanControl": "tune",
"ScanMetadata": "list_alt",
"ScanProgressBar": "timelapse",
"ScatterWaveform": "scatter_plot",
"SignalComboBox": "list_alt",
"SignalLabel": "scoreboard",
"SignalLineEdit": "vital_signs",
"SpinnerWidget": "progress_activity",
"StopButton": "dangerous",
"TextBox": "chat",
"ToggleSwitch": "toggle_on",
"Waveform": "show_chart",
"WebsiteWidget": "travel_explore",
"WidgetFinderComboBox": "frame_inspect",
}
+4 -7
View File
@@ -248,9 +248,7 @@ class RPCBase:
self._rpc_response = None
self._msg_wait_event.clear()
self._client.connector.register(
MessageEndpoints.gui_instruction_response(request_id),
cb=self._on_rpc_response,
parent=self,
MessageEndpoints.gui_instruction_response(request_id), cb=self._on_rpc_response
)
self._client.connector.set_and_publish(MessageEndpoints.gui_instructions(receiver), rpc_msg)
@@ -276,11 +274,10 @@ class RPCBase:
self._rpc_response = None
return self._create_widget_from_msg_result(msg_result)
@staticmethod
def _on_rpc_response(msg_obj: MessageObject, parent: RPCBase) -> None:
def _on_rpc_response(self, msg_obj: MessageObject) -> None:
msg = cast(messages.RequestResponseMessage, msg_obj.value)
parent._rpc_response = msg
parent._msg_wait_event.set()
self._rpc_response = msg
self._msg_wait_event.set()
def _create_widget_from_msg_result(self, msg_result):
if msg_result is None:
-57
View File
@@ -1,57 +0,0 @@
from __future__ import annotations
from bec_widgets.cli.client_utils import IGNORE_WIDGETS
from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.plugin_utils import get_custom_classes
class RPCWidgetHandler:
"""Handler class for creating widgets from RPC messages."""
def __init__(self):
self._widget_classes = None
@property
def widget_classes(self) -> dict[str, type[BECWidget]]:
"""
Get the available widget classes.
Returns:
dict: The available widget classes.
"""
if self._widget_classes is None:
self.update_available_widgets()
return self._widget_classes # type: ignore
def update_available_widgets(self):
"""
Update the available widgets.
Returns:
None
"""
self._widget_classes = (
get_custom_classes("bec_widgets", packages=("widgets", "applications"))
+ get_all_plugin_widgets()
).as_dict(IGNORE_WIDGETS)
def create_widget(self, widget_type, **kwargs) -> BECWidget:
"""
Create a widget from an RPC message.
Args:
widget_type(str): The type of the widget.
name (str): The name of the widget.
**kwargs: The keyword arguments for the widget.
Returns:
widget(BECWidget): The created widget.
"""
widget_class = self.widget_classes.get(widget_type) # type: ignore
if widget_class:
return widget_class(**kwargs)
raise ValueError(f"Unknown widget type: {widget_type}")
widget_handler = RPCWidgetHandler()
@@ -25,8 +25,8 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.rpc_widget_handler import widget_handler
from bec_widgets.utils.widget_io import WidgetHierarchy as wh
from bec_widgets.widgets.editors.jupyter_console.jupyter_console import BECJupyterConsole
@@ -206,7 +206,6 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
def _populate_registry_widgets(self):
try:
widget_handler.update_available_widgets()
items = sorted(widget_handler.widget_classes.keys())
except Exception as exc:
print(f"Failed to load registered widgets: {exc}")
@@ -335,20 +334,13 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
If kwargs does not contain `object_name`, it will default to the provided shortcut.
"""
# Ensure registry is loaded
widget_handler.update_available_widgets()
cls = widget_handler.widget_classes.get(widget_type)
if cls is None:
raise ValueError(f"Unknown registered widget type: {widget_type}")
if kwargs is None:
kwargs = {"object_name": shortcut}
else:
kwargs = dict(kwargs)
kwargs.setdefault("object_name", shortcut)
# Instantiate and add
widget = cls(**kwargs)
widget = widget_handler.create_widget(widget_type, **kwargs)
if not isinstance(widget, QWidget):
raise TypeError(
f"Instantiated object for type '{widget_type}' is not a QWidget: {type(widget)}"
-12
View File
@@ -1,13 +1 @@
from qtpy.QtWebEngineWidgets import QWebEngineView
from .bec_connector import BECConnector, ConnectionConfig
from .bec_dispatcher import BECDispatcher
from .bec_table import BECTable
from .colors import Colors
from .container_utils import WidgetContainerUtils
from .crosshair import Crosshair
from .entry_validator import EntryValidator
from .layout_manager import GridLayoutManager
from .rpc_decorator import register_rpc_methods, rpc_public
from .ui_loader import UILoader
from .validator_delegate import DoubleValidationDelegate
+1 -1
View File
@@ -15,9 +15,9 @@ from pydantic import BaseModel, Field, field_validator
from qtpy.QtCore import Property, QObject, QRunnable, QThreadPool, Signal
from qtpy.QtWidgets import QApplication
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.error_popups import ErrorPopupUtility, SafeSlot
from bec_widgets.utils.name_utils import sanitize_namespace
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.widget_io import WidgetHierarchy
from bec_widgets.utils.yaml_dialog import load_yaml, load_yaml_gui, save_yaml, save_yaml_gui
+9 -6
View File
@@ -175,12 +175,15 @@ class BECDispatcher:
cb_info (dict | None): A dictionary containing information about the callback. Defaults to None.
"""
qt_slot = QtThreadSafeCallback(cb=slot, cb_info=cb_info)
if qt_slot not in self._registered_slots:
self._registered_slots[qt_slot] = qt_slot
qt_slot = self._registered_slots[qt_slot]
self.client.connector.register(topics, cb=qt_slot, **kwargs)
topics_str, _ = self.client.connector._convert_endpointinfo(topics)
qt_slot.topics.update(set(topics_str))
if not self.client.connector.any_stream_is_registered(topics, qt_slot):
if qt_slot not in self._registered_slots:
self._registered_slots[qt_slot] = qt_slot
qt_slot = self._registered_slots[qt_slot]
self.client.connector.register(topics, cb=qt_slot, **kwargs)
topics_str, _ = self.client.connector._convert_endpointinfo(topics)
qt_slot.topics.update(set(topics_str))
else:
logger.warning(f"Attempted to create duplicate stream subscription for {topics=}")
def disconnect_slot(
self, slot: Callable, topics: EndpointInfo | str | list[EndpointInfo] | list[str]
+56 -4
View File
@@ -4,6 +4,7 @@ import importlib.metadata
import inspect
import pkgutil
import traceback
from functools import lru_cache
from importlib import util as importlib_util
from importlib.machinery import FileFinder, ModuleSpec, SourceFileLoader
from types import ModuleType
@@ -11,7 +12,11 @@ from typing import Generator
from bec_lib.logger import bec_logger
from bec_widgets.utils.plugin_utils import BECClassContainer, BECClassInfo
from bec_widgets.utils.plugin_utils import (
BECClassContainer,
BECClassInfo,
rpc_widget_registry_from_source,
)
logger = bec_logger.logger
@@ -53,6 +58,14 @@ def _submodule_by_name(module: ModuleType, name: str):
return None
def _submodule_spec_by_name(module: ModuleType, name: str) -> ModuleSpec | None:
for module_info in pkgutil.iter_modules(module.__path__):
if module_info.name != name or not isinstance(module_info.module_finder, FileFinder):
continue
return module_info.module_finder.find_spec(module_info.name)
return None
def _get_widgets_from_module(module: ModuleType) -> BECClassContainer:
"""Find any BECWidget subclasses in the given module and return them with their info."""
from bec_widgets.utils.bec_widget import BECWidget # avoid circular import
@@ -90,16 +103,55 @@ def get_plugin_client_module() -> ModuleType | None:
return _submodule_by_name(plugin, "client") if (plugin := user_widget_plugin()) else None
def get_plugin_designer_module() -> ModuleType | None:
"""If there is a plugin repository installed, return the designer module."""
return (
_submodule_by_name(plugin, "designer_plugins") if (plugin := user_widget_plugin()) else None
)
@lru_cache
def get_plugin_rpc_widget_registry() -> dict[str, tuple[str, str]]:
"""If there is a plugin repository installed, return the RPC widget registry."""
plugin = user_widget_plugin()
if plugin is None:
return {}
client_spec = _submodule_spec_by_name(plugin, "client")
if client_spec is not None and client_spec.origin:
try:
return rpc_widget_registry_from_source(client_spec.origin)
except (OSError, SyntaxError) as exc:
logger.warning(f"Could not parse plugin RPC widget registry: {exc}")
client_module = get_plugin_client_module()
if client_module is None:
return {}
registry = {}
for plugin_name, plugin_class in inspect.getmembers(client_module, inspect.isclass):
if hasattr(plugin_class, "_IMPORT_MODULE"):
registry[plugin_name] = (plugin_class._IMPORT_MODULE, plugin_class.__name__)
return registry
@lru_cache
def get_plugin_designer_registry() -> dict[str, tuple[str, str]]:
"""If there is a plugin repository installed, return the designer plugin registry."""
designer_module = get_plugin_designer_module()
if designer_module and hasattr(designer_module, "designer_plugins"):
return designer_module.designer_plugins
return {}
def get_all_plugin_widgets() -> BECClassContainer:
"""If there is a plugin repository installed, load all widgets from it."""
if plugin := user_widget_plugin():
return _all_widgets_from_all_submods(plugin)
else:
return BECClassContainer()
return BECClassContainer()
if __name__ == "__main__": # pragma: no cover
widgets = get_plugin_rpc_widget_registry()
client = get_plugin_client_module()
print(get_all_plugin_widgets())
...
+1 -1
View File
@@ -10,11 +10,11 @@ from qtpy.QtGui import QFont, QPixmap
from qtpy.QtWidgets import QApplication, QFileDialog, QLabel, QVBoxLayout, QWidget
import bec_widgets.widgets.containers.qt_ads as QtAds
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.busy_loader import install_busy_loader
from bec_widgets.utils.error_popups import SafeConnect, SafeSlot
from bec_widgets.utils.rpc_decorator import rpc_timeout
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.widget_io import WidgetHierarchy
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
@@ -7,31 +7,22 @@ import inspect
import os
import sys
from pathlib import Path
from typing import get_overloads
import black
import isort
from bec_lib.logger import bec_logger
from qtpy.QtCore import Property as QtProperty
from bec_widgets.utils.generate_designer_plugin import DesignerPluginGenerator, plugin_filenames
from bec_widgets.utils.generate_designer_plugin import (
DesignerPluginGenerator,
DesignerPluginInfo,
plugin_filenames,
)
from bec_widgets.utils.plugin_utils import BECClassContainer, get_custom_classes
logger = bec_logger.logger
if sys.version_info >= (3, 11):
from typing import get_overloads
else:
print(
"Python version is less than 3.11, using dummy function for get_overloads. "
"If you want to use the real function 'typing.get_overloads()', please use Python 3.11 or later."
)
def get_overloads(_obj):
"""
Dummy function for Python versions before 3.11.
"""
return []
class ClientGenerator:
def __init__(self, base=False):
@@ -54,7 +45,7 @@ from __future__ import annotations
from bec_lib.logger import bec_logger
from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call, rpc_timeout
{"from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets, get_plugin_client_module" if self._base else ""}
{"from bec_widgets.utils.bec_plugin_helper import get_plugin_client_module" if self._base else ""}
logger = bec_logger.logger
@@ -111,27 +102,19 @@ _Widgets = {
self.content += """
try:
_plugin_widgets = get_all_plugin_widgets().as_dict()
plugin_client = get_plugin_client_module()
Widgets = _WidgetsEnumType("Widgets", {name: name for name in _plugin_widgets} | _Widgets)
if (_overlap := _Widgets.keys() & _plugin_widgets.keys()) != set():
for _widget in _overlap:
logger.warning(f"Detected duplicate widget {_widget} in plugin repo file: {inspect.getfile(_plugin_widgets[_widget])} !")
for plugin_name, plugin_class in inspect.getmembers(plugin_client, inspect.isclass):
if issubclass(plugin_class, RPCBase) and plugin_class is not RPCBase:
if plugin_name not in _Widgets:
_Widgets[plugin_name] = plugin_name
if plugin_name in globals():
conflicting_file = (
inspect.getfile(_plugin_widgets[plugin_name])
if plugin_name in _plugin_widgets
else f"{plugin_client}"
)
logger.warning(
f"Plugin widget {plugin_name} from {conflicting_file} conflicts with a built-in class!"
f"Plugin widget {plugin_name} in {plugin_class._IMPORT_MODULE} conflicts with a built-in class!"
)
continue
if plugin_name not in _overlap:
else:
globals()[plugin_name] = plugin_class
Widgets = _WidgetsEnumType("Widgets", _Widgets)
except ImportError as e:
logger.error(f"Failed loading plugins: \\n{reduce(add, traceback.format_exception(e))}")
"""
@@ -146,12 +129,8 @@ except ImportError as e:
class_name = cls.__name__
if class_name == "BECDockArea":
self.content += f"""
class {class_name}(RPCBase):"""
else:
self.content += f"""
class {class_name}(RPCBase):"""
self.content += f"""
class {class_name}(RPCBase):\n"""
if cls.__doc__:
# We only want the first line of the docstring
@@ -162,13 +141,9 @@ class {class_name}(RPCBase):"""
else:
class_docs = cls.__doc__.split("\n")[1]
self.content += f"""
\"\"\"{class_docs}\"\"\"
"""
\"\"\"{class_docs}\"\"\"\n"""
user_access_entries = self._get_user_access_entries(cls)
if not user_access_entries:
self.content += """...
"""
self.content += f' _IMPORT_MODULE="{cls.__module__}"\n'
for method_entry in user_access_entries:
method, obj, is_property_setter = self._resolve_method_object(cls, method_entry)
if obj is None:
@@ -279,6 +254,58 @@ class {class_name}(RPCBase):"""
file.write(formatted_content)
def write_designer_plugins(plugin_infos: list[DesignerPluginInfo], file_name: str):
"""
Write a registry of Qt widget classes with designer plugins.
Args:
plugin_infos(list[DesignerPluginInfo]): The designer plugin metadata to write.
file_name(str): The name of the file to write to.
"""
plugin_infos = sorted(plugin_infos, key=lambda info: info.plugin_name_pascal)
content = """# This file was automatically generated by generate_cli.py
# type: ignore
from __future__ import annotations
# pylint: skip-file
designer_plugins = {
"""
for info in plugin_infos:
widget_module = info.plugin_class.__module__
widget_class = info.plugin_name_pascal
content += f' "{info.plugin_name_pascal}": ("{widget_module}", "{widget_class}"),\n'
content += """
}
widget_icons = {
"""
for info in plugin_infos:
content += f' "{info.plugin_name_pascal}": "{info.icon_name}",\n'
content += """
}
"""
try:
formatted_content = black.format_str(content, mode=black.Mode(line_length=100))
except black.NothingChanged:
formatted_content = content
config = isort.Config(
profile="black",
line_length=100,
multi_line_output=3,
include_trailing_comma=False,
known_first_party=["bec_widgets"],
)
formatted_content = isort.code(formatted_content, config=config)
with open(file_name, "w", encoding="utf-8") as file:
file.write(formatted_content)
def main():
"""
Main entry point for the script, controlled by command line arguments.
@@ -332,6 +359,8 @@ def main():
else:
non_overwrite_classes = []
designer_plugin_infos = []
for cls in rpc_classes.plugins:
logger.info(f"Writing bec-designer plugin files for {cls.__name__}...")
@@ -339,21 +368,30 @@ def main():
logger.error(
f"Not writing plugin files for {cls.__name__} because a built-in widget with that name exists"
)
continue
plugin = DesignerPluginGenerator(cls)
if not hasattr(plugin, "info"):
if not hasattr(plugin, "info") or plugin.excluded:
continue
def _exists(file: str):
return os.path.exists(os.path.join(plugin.info.base_path, file))
if any(_exists(file) for file in plugin_filenames(plugin.info.plugin_name_snake)):
if _exists(plugin.filenames.plugin):
designer_plugin_infos.append(plugin.info)
logger.debug(
f"Skipping generation of extra plugin files for {plugin.info.plugin_name_snake} - at least one file out of 'plugin.py', 'pyproject', and 'register_{plugin.info.plugin_name_snake}.py' already exists."
)
continue
plugin.run()
designer_plugin_infos.append(plugin.info)
# Write designer_plugins.py with plugin import metadata for all widgets with designer plugins.
designer_plugins_path = module_dir / client_subdir / "designer_plugins.py"
logger.info(f"Generating designer plugin registry at {designer_plugins_path}")
write_designer_plugins(designer_plugin_infos, str(designer_plugins_path))
if __name__ == "__main__": # pragma: no cover
@@ -29,6 +29,7 @@ class DesignerPluginInfo:
self.plugin_name_pascal = plugin_class.__name__
self.plugin_name_snake = pascal_to_snake(self.plugin_name_pascal)
self.widget_import = f"from {plugin_class.__module__} import {self.plugin_name_pascal}"
self.icon_name = getattr(plugin_class, "ICON_NAME", "")
plugin_module = (
".".join(plugin_class.__module__.split(".")[:-1]) + f".{self.plugin_name_snake}_plugin"
)
@@ -63,6 +64,10 @@ class DesignerPluginGenerator:
def filenames(self):
return plugin_filenames(self.info.plugin_name_snake)
@property
def excluded(self):
return self._excluded
def run(self, validate=True):
if self._excluded:
print(f"Plugin {self.widget.__name__} is excluded from generation.")
+110 -50
View File
@@ -1,56 +1,22 @@
from __future__ import annotations
import ast
import importlib
import inspect
import os
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING, Iterable
from bec_lib.plugin_helper import _get_available_plugins
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_widget import BECWidget
if TYPE_CHECKING: # pragma: no cover
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates
def get_plugin_widgets() -> dict[str, BECConnector]:
"""
Get all available widgets from the plugin directory. Widgets are classes that inherit from BECConnector.
The plugins are provided through python plugins and specified in the respective pyproject.toml file using
the following key:
[project.entry-points."bec.widgets.user_widgets"]
plugin_widgets = "path.to.plugin.module"
e.g.
[project.entry-points."bec.widgets.user_widgets"]
plugin_widgets = "pxiii_bec.bec_widgets.widgets"
assuming that the widgets module for the package pxiii_bec is located at pxiii_bec/bec_widgets/widgets and
contains the widgets to be loaded within the pxiii_bec/bec_widgets/widgets/__init__.py file.
Returns:
dict[str, BECConnector]: A dictionary of widget names and their respective classes.
"""
modules = _get_available_plugins("bec.widgets.user_widgets")
loaded_plugins = {}
print(modules)
for module in modules:
mods = inspect.getmembers(module, predicate=_filter_plugins)
for name, mod_cls in mods:
if name in loaded_plugins:
print(f"Duplicated widgets plugin {name}.")
loaded_plugins[name] = mod_cls
return loaded_plugins
def _filter_plugins(obj):
return inspect.isclass(obj) and issubclass(obj, BECConnector)
def get_plugin_auto_updates() -> dict[str, type[AutoUpdates]]:
"""
Get all available auto update classes from the plugin directory. AutoUpdates must inherit from AutoUpdate and be
@@ -66,6 +32,8 @@ def get_plugin_auto_updates() -> dict[str, type[AutoUpdates]]:
Returns:
dict[str, AutoUpdates]: A dictionary of widget names and their respective classes.
"""
from bec_lib.plugin_helper import _get_available_plugins
modules = _get_available_plugins("bec.widgets.auto_updates")
loaded_plugins = {}
for module in modules:
@@ -168,6 +136,11 @@ class BECClassContainer:
def _collect_classes_from_package(repo_name: str, package: str) -> BECClassContainer:
"""Collect classes from a package subtree (for example ``widgets`` or ``applications``)."""
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.bec_widget import BECWidget
collection = BECClassContainer()
try:
anchor_module = importlib.import_module(f"{repo_name}.{package}")
@@ -194,17 +167,18 @@ def _collect_classes_from_package(repo_name: str, package: str) -> BECClassConta
for name in dir(module):
obj = getattr(module, name)
if not isinstance(obj, type):
continue
if not hasattr(obj, "__module__") or obj.__module__ != module.__name__:
continue
if isinstance(obj, type):
class_info = BECClassInfo(name=name, module=module.__name__, file=path, obj=obj)
if issubclass(obj, BECConnector):
class_info.is_connector = True
if issubclass(obj, QWidget) or issubclass(obj, BECWidget):
class_info.is_widget = True
if hasattr(obj, "PLUGIN") and obj.PLUGIN:
class_info.is_plugin = True
collection.add_class(class_info)
class_info = BECClassInfo(name=name, module=module.__name__, file=path, obj=obj)
if issubclass(obj, BECConnector):
class_info.is_connector = True
if issubclass(obj, QWidget) or issubclass(obj, BECWidget):
class_info.is_widget = True
if hasattr(obj, "PLUGIN") and obj.PLUGIN:
class_info.is_plugin = True
collection.add_class(class_info)
return collection
@@ -229,3 +203,89 @@ def get_custom_classes(
for package in selected_packages:
collection += _collect_classes_from_package(repo_name, package)
return collection
def _get_designer_registry() -> dict[str, tuple[str, str]]:
from bec_widgets.cli.designer_plugins import designer_plugins
return designer_plugins
def _resolve_widget_from_registry(import_path: str, widget_name: str) -> type[QWidget]:
widget = importlib.import_module(import_path)
return getattr(widget, widget_name)
def designer_plugin_exists(name: str) -> bool:
from bec_widgets.utils.bec_plugin_helper import get_plugin_designer_registry
internal_registry = _get_designer_registry()
external_registry = get_plugin_designer_registry()
return name in internal_registry or name in external_registry
def get_designer_plugin(name: str, raise_on_missing: bool = True) -> type[QWidget] | None:
from bec_widgets.utils.bec_plugin_helper import get_plugin_designer_registry
internal_registry = _get_designer_registry()
external_registry = get_plugin_designer_registry()
if name in external_registry:
import_path, widget_name = external_registry[name]
return _resolve_widget_from_registry(import_path, widget_name)
if name in internal_registry:
import_path, widget_name = internal_registry[name]
return _resolve_widget_from_registry(import_path, widget_name)
if raise_on_missing:
raise ValueError(
f"Designer plugin {name} not found in either internal or external registry."
)
return None
def rpc_widget_registry_from_source(path: str | Path) -> dict[str, tuple[str, str]]:
"""Parse a generated RPC client module and return its widget registry."""
source_path = Path(path)
module_node = ast.parse(source_path.read_text(encoding="utf-8"), filename=str(source_path))
registry = {}
for node in module_node.body:
if not isinstance(node, ast.ClassDef):
continue
for item in node.body:
if not isinstance(item, ast.Assign):
continue
if not any(
isinstance(target, ast.Name) and target.id == "_IMPORT_MODULE"
for target in item.targets
):
continue
if isinstance(item.value, ast.Constant) and isinstance(item.value.value, str):
registry[node.name] = (item.value.value, node.name)
break
return registry
@lru_cache
def get_rpc_widget_registry() -> dict[str, tuple[str, str]]:
client_path = Path(__file__).resolve().parents[1] / "cli" / "client.py"
return rpc_widget_registry_from_source(client_path)
@lru_cache
def rpc_widget_registry() -> dict[str, tuple[str, str]]:
from bec_widgets.utils.bec_plugin_helper import get_plugin_rpc_widget_registry
internal_registry = get_rpc_widget_registry()
external_registry = get_plugin_rpc_widget_registry()
return {**external_registry, **internal_registry}
def get_rpc_widget(name: str, raise_on_missing: bool = True) -> type[QWidget] | None:
registry = rpc_widget_registry()
if name in registry:
import_path, widget_name = registry[name]
return _resolve_widget_from_registry(import_path, widget_name)
if raise_on_missing:
raise ValueError(f"RPC widget {name} not found in registry.")
return None
+2 -2
View File
@@ -14,11 +14,11 @@ from qtpy.QtCore import Qt, QTimer
from qtpy.QtWidgets import QWidget
from redis.exceptions import RedisError
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import BECDispatcher
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.error_popups import ErrorPopupUtility
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.screen_utils import apply_window_geometry
from bec_widgets.widgets.containers.dock_area.dock_area import BECDockArea
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow, BECMainWindowNoRPC
+49
View File
@@ -0,0 +1,49 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from bec_widgets.utils.plugin_utils import get_rpc_widget, rpc_widget_registry
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.utils.bec_widget import BECWidget
class RPCWidgetHandler:
"""Handler class for creating widgets from RPC messages."""
def __init__(self):
self._widget_registry = None
@property
def widget_classes(self) -> dict[str, tuple[str, str]]:
"""
Get the available widget classes.
Returns:
dict: The available widget classes.
"""
registry = rpc_widget_registry()
if not registry:
return {}
return registry
@staticmethod
def create_widget(widget_type, **kwargs) -> BECWidget:
"""
Create a widget from an RPC message.
Args:
widget_type(str): The type of the widget.
name (str): The name of the widget.
**kwargs: The keyword arguments for the widget.
Returns:
widget(BECWidget): The created widget.
"""
widget = get_rpc_widget(widget_type, raise_on_missing=False)
if widget:
return widget(**kwargs)
raise ValueError(f"Unknown widget type: {widget_type}")
widget_handler = RPCWidgetHandler()
+9 -85
View File
@@ -1,10 +1,8 @@
from bec_lib.logger import bec_logger
from qtpy import PYQT6, PYSIDE6
from qtpy import PYSIDE6
from qtpy.QtCore import QFile, QIODevice
from bec_widgets.utils.bec_plugin_helper import get_all_plugin_widgets
from bec_widgets.utils.generate_designer_plugin import DesignerPluginInfo
from bec_widgets.utils.plugin_utils import get_custom_classes
from bec_widgets.utils.plugin_utils import get_designer_plugin
logger = bec_logger.logger
@@ -12,16 +10,14 @@ if PYSIDE6:
from qtpy.QtUiTools import QUiLoader
class CustomUiLoader(QUiLoader):
def __init__(self, baseinstance, custom_widgets: dict | None = None):
def __init__(self, baseinstance):
super().__init__(baseinstance)
self.custom_widgets = custom_widgets or {}
self.baseinstance = baseinstance
def createWidget(self, class_name, parent=None, name=""):
if class_name in self.custom_widgets:
widget = self.custom_widgets[class_name](self.baseinstance)
return widget
widget = get_designer_plugin(class_name, raise_on_missing=False)
if widget is not None:
return widget(self.baseinstance)
return super().createWidget(class_name, self.baseinstance, name)
@@ -31,16 +27,9 @@ class UILoader:
def __init__(self, parent=None):
self.parent = parent
self.custom_widgets = (
get_custom_classes("bec_widgets") + get_all_plugin_widgets()
).as_dict()
if PYSIDE6:
self.loader = self.load_ui_pyside6
elif PYQT6:
self.loader = self.load_ui_pyqt6
else:
if not PYSIDE6:
raise ImportError("No compatible Qt bindings found.")
self.loader = self.load_ui_pyside6
def load_ui_pyside6(self, ui_file, parent=None):
"""
@@ -53,7 +42,7 @@ class UILoader:
QWidget: The loaded widget.
"""
parent = parent or self.parent
loader = CustomUiLoader(parent, self.custom_widgets)
loader = CustomUiLoader(parent)
file = QFile(ui_file)
if not file.open(QIODevice.ReadOnly):
raise IOError(f"Cannot open file: {ui_file}")
@@ -61,71 +50,6 @@ class UILoader:
file.close()
return widget
def load_ui_pyqt6(self, ui_file, parent=None):
"""
Specific loader for PyQt6 using loadUi.
Args:
ui_file(str): Path to the .ui file.
parent(QWidget): Parent widget.
Returns:
QWidget: The loaded widget.
"""
from PyQt6.uic.Loader.loader import DynamicUILoader
class CustomDynamicUILoader(DynamicUILoader):
def __init__(self, package, custom_widgets: dict = None):
super().__init__(package)
self.custom_widgets = custom_widgets or {}
def _handle_custom_widgets(self, el):
"""Handle the <customwidgets> element."""
def header2module(header):
"""header2module(header) -> string
Convert paths to C++ header files to according Python modules
>>> header2module("foo/bar/baz.h")
'foo.bar.baz'
"""
if header.endswith(".h"):
header = header[:-2]
mpath = []
for part in header.split("/"):
# Ignore any empty parts or those that refer to the current
# directory.
if part not in ("", "."):
if part == "..":
# We should allow this for Python3.
raise SyntaxError(
"custom widget header file name may not contain '..'."
)
mpath.append(part)
return ".".join(mpath)
for custom_widget in el:
classname = custom_widget.findtext("class")
header = custom_widget.findtext("header")
if header:
header = self._translate_bec_widgets_header(header)
self.factory.addCustomWidget(
classname,
custom_widget.findtext("extends") or "QWidget",
header2module(header),
)
def _translate_bec_widgets_header(self, header):
for name, value in self.custom_widgets.items():
if header == DesignerPluginInfo.pascal_to_snake(name):
return value.__module__
return header
return CustomDynamicUILoader("", self.custom_widgets).loadUi(ui_file, parent)
def load_ui(self, ui_file, parent=None):
"""
Universal UI loader method.
+6 -6
View File
@@ -26,7 +26,7 @@ from qtpy.QtWidgets import (
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
logger = bec_logger.logger
@@ -418,7 +418,7 @@ class WidgetHierarchy:
only_bec_widgets(bool, optional): Whether to print only widgets that are instances of BECWidget.
show_parent(bool, optional): Whether to display which BECWidget is the parent of each discovered BECWidget.
"""
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.widgets.plots.waveform.waveform import Waveform
for node in WidgetHierarchy.iter_widget_tree(
@@ -468,7 +468,7 @@ class WidgetHierarchy:
from qtpy.QtWidgets import QApplication
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.widgets.plots.plot_base import PlotBase
# 1) Gather ALL QWidget-based BECConnector objects
@@ -534,7 +534,7 @@ class WidgetHierarchy:
Returns:
The nearest ancestor that is a BECConnector, or None if not found.
"""
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
# Guard against deleted/invalid Qt wrappers
if not shb.isValid(widget):
@@ -636,7 +636,7 @@ class WidgetHierarchy:
Return all BECConnector instances whose closest BECConnector ancestor is the given widget,
including the widget itself if it is a BECConnector.
"""
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
connectors: list[BECConnector] = []
if isinstance(widget, BECConnector):
@@ -664,7 +664,7 @@ class WidgetHierarchy:
return None
try:
from bec_widgets.utils import BECConnector # local import to avoid cycles
from bec_widgets.utils.bec_connector import BECConnector # local import to avoid cycles
is_bec_target = False
if isinstance(ancestor_class, str):
@@ -13,9 +13,9 @@ from shiboken6 import isValid
import bec_widgets.widgets.containers.qt_ads as QtAds
from bec_widgets import BECWidget, SafeSlot
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.property_editor import PropertyEditor
from bec_widgets.utils.rpc_widget_handler import widget_handler
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.widgets.containers.qt_ads import (
CDockAreaWidget,
@@ -19,11 +19,10 @@ from qtpy.QtWidgets import (
import bec_widgets.widgets.containers.qt_ads as QtAds
from bec_widgets import BECWidget, SafeProperty, SafeSlot
from bec_widgets.applications.views.view import ViewTourSteps
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils import BECDispatcher
from bec_widgets.cli.designer_plugins import widget_icons
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.rpc_decorator import rpc_timeout
from bec_widgets.utils.rpc_widget_handler import widget_handler
from bec_widgets.utils.toolbars.actions import (
ExpandableMenuAction,
MaterialIconAction,
@@ -35,25 +34,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 (
@@ -65,22 +64,7 @@ from bec_widgets.widgets.containers.dock_area.toolbar_components.workspace_actio
WorkspaceConnection,
workspace_bundle,
)
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindowNoRPC
from bec_widgets.widgets.containers.qt_ads import CDockWidget
from bec_widgets.widgets.control.device_control.positioner_box import PositionerBox, PositionerBox2D
from bec_widgets.widgets.control.scan_control import ScanControl
from bec_widgets.widgets.editors.bec_console.bec_console import BecConsole, BECShell
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.services.bec_queue.bec_queue import BECQueue
from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox
from bec_widgets.widgets.utility.logpanel import LogPanel
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
logger = bec_logger.logger
@@ -108,6 +92,7 @@ class BECDockArea(DockAreaWidget):
"list_profiles",
"save_profile",
"load_profile",
"restore_baseline_profile",
"delete_profile",
]
@@ -143,6 +128,10 @@ class BECDockArea(DockAreaWidget):
self._mode = mode
# Toolbar
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import (
DarkModeButton,
)
self.dark_mode_button = DarkModeButton(parent=self, toolbar=True)
self.dark_mode_button.setVisible(enable_profile_management)
self._setup_toolbar()
@@ -235,11 +224,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:
"""
@@ -344,38 +330,42 @@ class BECDockArea(DockAreaWidget):
self.toolbar = ModularToolBar(parent=self)
plot_actions = {
"waveform": (Waveform.ICON_NAME, "Add Waveform", "Waveform"),
"waveform": (widget_icons["Waveform"], "Add Waveform", "Waveform"),
"scatter_waveform": (
ScatterWaveform.ICON_NAME,
widget_icons["ScatterWaveform"],
"Add Scatter Waveform",
"ScatterWaveform",
),
"multi_waveform": (MultiWaveform.ICON_NAME, "Add Multi Waveform", "MultiWaveform"),
"image": (Image.ICON_NAME, "Add Image", "Image"),
"motor_map": (MotorMap.ICON_NAME, "Add Motor Map", "MotorMap"),
"heatmap": (Heatmap.ICON_NAME, "Add Heatmap", "Heatmap"),
"multi_waveform": (
widget_icons["MultiWaveform"],
"Add Multi Waveform",
"MultiWaveform",
),
"image": (widget_icons["Image"], "Add Image", "Image"),
"motor_map": (widget_icons["MotorMap"], "Add Motor Map", "MotorMap"),
"heatmap": (widget_icons["Heatmap"], "Add Heatmap", "Heatmap"),
}
device_actions = {
"scan_control": (ScanControl.ICON_NAME, "Add Scan Control", "ScanControl"),
"positioner_box": (PositionerBox.ICON_NAME, "Add Device Box", "PositionerBox"),
"scan_control": (widget_icons["ScanControl"], "Add Scan Control", "ScanControl"),
"positioner_box": (widget_icons["PositionerBox"], "Add Device Box", "PositionerBox"),
"positioner_box_2D": (
PositionerBox2D.ICON_NAME,
widget_icons["PositionerBox2D"],
"Add Device 2D Box",
"PositionerBox2D",
),
}
util_actions = {
"queue": (BECQueue.ICON_NAME, "Add Scan Queue", "BECQueue"),
"status": (BECStatusBox.ICON_NAME, "Add BEC Status Box", "BECStatusBox"),
"queue": (widget_icons["BECQueue"], "Add Scan Queue", "BECQueue"),
"status": (widget_icons["BECStatusBox"], "Add BEC Status Box", "BECStatusBox"),
"progress_bar": (
RingProgressBar.ICON_NAME,
widget_icons["RingProgressBar"],
"Add Circular ProgressBar",
"RingProgressBar",
),
"terminal": (BecConsole.ICON_NAME, "Add Terminal", "BecConsole"),
"bec_shell": (BECShell.ICON_NAME, "Add BEC Shell", "BECShell"),
"log_panel": (LogPanel.ICON_NAME, "Add LogPanel - Disabled", "LogPanel"),
"sbb_monitor": ("train", "Add SBB Monitor", "SBBMonitor"),
"terminal": (widget_icons["BecConsole"], "Add Terminal", "BecConsole"),
"bec_shell": (widget_icons["BECShell"], "Add BEC Shell", "BECShell"),
"sbb_monitor": (widget_icons["SBBMonitor"], "Add SBB Monitor", "SBBMonitor"),
"log_panel": (widget_icons["LogPanel"], "Add LogPanel", "LogPanel"),
}
# Create expandable menu actions (original behavior)
@@ -487,9 +477,7 @@ class BECDockArea(DockAreaWidget):
# first two items not needed for this part
for key, (_, _, widget_type) in mapping.items():
act = menu.actions[key].action
if widget_type == "LogPanel":
act.setEnabled(False) # keep disabled per issue #644
elif key == "terminal":
if key == "terminal":
act.triggered.connect(
lambda _, t=widget_type: self.new(widget=t, closable=True, startup_cmd=None)
)
@@ -510,10 +498,7 @@ class BECDockArea(DockAreaWidget):
for action_id, (_, _, widget_type) in mapping.items():
flat_action_id = f"flat_{action_id}"
flat_action = self.toolbar.components.get_action(flat_action_id).action
if widget_type == "LogPanel":
flat_action.setEnabled(False) # keep disabled per issue #644
else:
flat_action.triggered.connect(lambda _, t=widget_type: self.new(t))
flat_action.triggered.connect(lambda _, t=widget_type: self.new(t))
_connect_flat_actions(self._ACTION_MAPPINGS["menu_plots"])
_connect_flat_actions(self._ACTION_MAPPINGS["menu_devices"])
@@ -595,13 +580,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:
"""
@@ -627,35 +612,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:
"""
@@ -673,6 +657,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.
@@ -709,10 +701,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:
@@ -776,7 +768,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
@@ -785,12 +777,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)
@@ -800,7 +792,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()
@@ -820,16 +811,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
@@ -848,14 +844,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()
@@ -897,32 +896,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)
@@ -1057,7 +1060,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)
@@ -1156,7 +1159,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
@@ -1186,6 +1189,8 @@ class BECDockArea(DockAreaWidget):
)
step_ids.append(step_id)
from bec_widgets.applications.views.view import ViewTourSteps
return ViewTourSteps(view_title="Dock Area Workspace", step_ids=step_ids)
def cleanup(self):
@@ -1206,6 +1211,9 @@ class BECDockArea(DockAreaWidget):
if __name__ == "__main__": # pragma: no cover
import sys
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindowNoRPC
app = QApplication(sys.argv)
apply_theme("dark")
dispatcher = BECDispatcher(gui_id="ads")
@@ -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)
@@ -22,7 +22,7 @@ from qtpy.QtWidgets import (
)
from typeguard import typechecked
from bec_widgets.cli.rpc.rpc_widget_handler import widget_handler
from bec_widgets.utils.rpc_widget_handler import widget_handler
class LayoutManagerWidget(QWidget):
@@ -28,7 +28,7 @@ from qtpy.QtCore import QObject, QTimer
from qtpy.QtWidgets import QApplication, QFrame, QMainWindow, QScrollArea, QWidget
from bec_widgets import SafeProperty, SafeSlot
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.widget_io import WidgetIO
@@ -18,10 +18,10 @@ from qtpy.QtWidgets import (
)
import bec_widgets
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.containers.main_window.addons.hover_widget import HoverWidget
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
BECNotificationBroker,
@@ -30,7 +30,6 @@ from bec_widgets.widgets.containers.main_window.addons.notification_center.notif
)
from bec_widgets.widgets.containers.main_window.addons.scroll_label import ScrollLabel
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
from bec_widgets.widgets.editors.bec_console.bec_console import BecConsoleRegistry
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import ScanProgressBar
from bec_widgets.widgets.utility.widget_hierarchy_tree.widget_hierarchy_tree import (
WidgetHierarchyDialog,
@@ -54,9 +53,6 @@ class BECMainWindow(BECWidget, QMainWindow):
super().__init__(parent=parent, **kwargs)
self.app = QApplication.instance()
self._console_registry = BecConsoleRegistry(self)
if not hasattr(self.app, "console_widget_registry"):
self.app.console_widget_registry = self._console_registry
self.status_bar = self.statusBar()
self._launcher_window = None
self.setWindowTitle(window_title)
@@ -11,9 +11,9 @@ from qtpy.QtCore import Qt, Signal
from qtpy.QtGui import QDoubleValidator
from qtpy.QtWidgets import QDoubleSpinBox
from bec_widgets.utils import UILoader
from bec_widgets.utils.colors import apply_theme, get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base import (
DeviceUpdateUIComponents,
PositionerBoxBase,
@@ -12,9 +12,9 @@ from qtpy.QtCore import Signal
from qtpy.QtGui import QDoubleValidator
from qtpy.QtWidgets import QDoubleSpinBox
from bec_widgets.utils import UILoader
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base import (
DeviceUpdateUIComponents,
PositionerBoxBase,
@@ -7,7 +7,7 @@ from bec_lib.device import Signal as BECSignal
from bec_lib.logger import bec_logger
from pydantic import field_validator
from bec_widgets.utils import ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.filter_io import FilterIO
@@ -3,7 +3,7 @@ from bec_lib.device import Signal
from bec_lib.logger import bec_logger
from qtpy.QtCore import Property
from bec_widgets.utils import ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.filter_io import FilterIO
@@ -19,7 +19,7 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.utils import ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme, get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
@@ -5,10 +5,10 @@ from bec_lib.logger import bec_logger
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QPushButton, QSizePolicy, QTreeWidgetItem, QVBoxLayout, QWidget
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ui_loader import UILoader
logger = bec_logger.logger
@@ -1,13 +1,13 @@
from __future__ import annotations
import enum
from dataclasses import dataclass, field
from uuid import uuid4
from weakref import WeakValueDictionary
import shiboken6
from bec_lib.logger import bec_logger
from pydantic import BaseModel
from qtpy.QtCore import QObject, Qt
from qtpy.QtCore import Qt, Signal
from qtpy.QtGui import QMouseEvent
from qtpy.QtWidgets import (
QApplication,
@@ -39,33 +39,133 @@ class ConsoleMode(str, enum.Enum):
HIDDEN = "hidden"
class _TerminalOwnerInfo(BaseModel):
@dataclass
class _TerminalOwnerInfo:
"""Should be managed only by the BecConsoleRegistry. Consoles should ask the registry for
necessary ownership info."""
owner_console_id: str | None = None
registered_console_ids: set[str] = set()
instance: BecTerminal | QWidget
terminal_id: str
registered_console_ids: set[str] = field(default_factory=set)
instance: BecTerminal | None = None
terminal_id: str = ""
initialized: bool = False
keep_if_last_console_closed: bool = False
model_config = {"arbitrary_types_allowed": True}
persist_session: bool = False
fallback_holder: QWidget | None = None
class BecConsoleRegistry(QWidget):
class BecConsoleRegistry:
"""
A registry for the BecConsole class to manage its instances.
"""
def __init__(self, parent):
def __init__(self):
"""
Initialize the registry.
"""
super().__init__(parent=parent)
self._consoles: WeakValueDictionary[str, BecConsole] = WeakValueDictionary()
self._terminal_registry: dict[str, _TerminalOwnerInfo] = {}
@staticmethod
def _is_valid_qobject(obj: object | None) -> bool:
return obj is not None and shiboken6.isValid(obj)
def _connect_app_cleanup(self) -> None:
app = QApplication.instance()
if app is None:
return
app.aboutToQuit.connect(self.clear, Qt.ConnectionType.UniqueConnection)
@staticmethod
def _new_terminal_info(console: BecConsole) -> _TerminalOwnerInfo:
term = _BecTermClass()
return _TerminalOwnerInfo(
registered_console_ids={console.console_id},
owner_console_id=console.console_id,
instance=term,
terminal_id=console.terminal_id,
persist_session=console.persist_terminal_session,
)
@staticmethod
def _replace_terminal(info: _TerminalOwnerInfo, console: BecConsole) -> None:
info.instance = _BecTermClass()
info.initialized = False
info.owner_console_id = console.console_id
info.registered_console_ids.add(console.console_id)
info.persist_session = info.persist_session or console.persist_terminal_session
def _delete_terminal_info(self, info: _TerminalOwnerInfo) -> None:
if self._is_valid_qobject(info.instance):
info.instance.deleteLater() # type: ignore[union-attr]
info.instance = None
if self._is_valid_qobject(info.fallback_holder):
info.fallback_holder.deleteLater()
info.fallback_holder = None
def _parking_parent(
self,
info: _TerminalOwnerInfo,
console: BecConsole | None = None,
*,
avoid_console: bool = False,
) -> QWidget | None:
for console_id in info.registered_console_ids:
candidate = self._consoles.get(console_id)
if candidate is None or candidate is console:
continue
if self._is_valid_qobject(candidate):
return candidate._term_holder
if console is None or not self._is_valid_qobject(console):
return None
window = console.window()
if window is not None and window is not console and self._is_valid_qobject(window):
return window
if not avoid_console:
return console._term_holder
return None
def _fallback_holder(
self,
info: _TerminalOwnerInfo,
console: BecConsole | None = None,
*,
avoid_console: bool = False,
) -> QWidget:
if not self._is_valid_qobject(info.fallback_holder):
info.fallback_holder = QWidget(
parent=self._parking_parent(info, console, avoid_console=avoid_console)
)
info.fallback_holder.setObjectName(f"_bec_console_terminal_holder_{info.terminal_id}")
info.fallback_holder.hide()
return info.fallback_holder
def _park_terminal(
self,
info: _TerminalOwnerInfo,
console: BecConsole | None = None,
*,
avoid_console: bool = False,
) -> None:
if not self._is_valid_qobject(info.instance):
return
parent = self._parking_parent(info, console, avoid_console=avoid_console)
if parent is None and info.persist_session:
parent = self._fallback_holder(info, console, avoid_console=avoid_console)
info.instance.hide() # type: ignore[union-attr]
info.instance.setParent(parent) # type: ignore[union-attr]
def clear(self) -> None:
"""Delete every tracked terminal and holder."""
for info in list(self._terminal_registry.values()):
self._delete_terminal_info(info)
self._terminal_registry.clear()
self._consoles.clear()
def register(self, console: BecConsole):
"""
Register an instance of BecConsole. If there is already a terminal with the associated
@@ -74,49 +174,56 @@ class BecConsoleRegistry(QWidget):
Args:
console (BecConsole): The instance to register.
"""
self._connect_app_cleanup()
self._consoles[console.console_id] = console
console_id, terminal_id = console.console_id, console.terminal_id
if (term_info := self._terminal_registry.get(terminal_id)) is None or not shiboken6.isValid(
term_info.instance
):
term = _BecTermClass()
self._terminal_registry[terminal_id] = _TerminalOwnerInfo(
registered_console_ids={console_id},
owner_console_id=console_id,
instance=term,
terminal_id=terminal_id,
keep_if_last_console_closed=console.persevere_terminal,
)
if console.persevere_terminal:
term.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose, False)
term_info = self._terminal_registry.get(terminal_id)
if term_info is None:
self._terminal_registry[terminal_id] = self._new_terminal_info(console)
return
logger.info(f"Registered new console {console_id} for terminal {terminal_id}")
term_info.persist_session = term_info.persist_session or console.persist_terminal_session
had_registered_consoles = bool(term_info.registered_console_ids)
term_info.registered_console_ids.add(console_id)
if not self._is_valid_qobject(term_info.instance):
self._replace_terminal(term_info, console)
return
if (
term_info.owner_console_id is not None
and term_info.owner_console_id not in self._consoles
):
term_info.owner_console_id = None
if term_info.owner_console_id is None and not had_registered_consoles:
term_info.owner_console_id = console_id
logger.info(f"Registered new console {console_id} for terminal {terminal_id}")
def unregister(self, console: BecConsole):
"""
Unregister an instance of BecConsole.
Args:
instance (BecConsole): The instance to unregister.
console (BecConsole): The instance to unregister.
"""
console_id, terminal_id = console.console_id, console.terminal_id
if console_id in self._consoles:
del self._consoles[console_id]
if (term_info := self._terminal_registry.get(terminal_id)) is None:
return
detached = console._detach_terminal_widget(term_info.instance)
if console_id in term_info.registered_console_ids:
term_info.registered_console_ids.remove(console_id)
if term_info.owner_console_id == console_id:
term_info.owner_console_id = None
if not term_info.registered_console_ids:
if not term_info.keep_if_last_console_closed:
term_info.instance.deleteLater()
del self._terminal_registry[terminal_id]
else:
term_info.instance.setHidden()
term_info.instance.setParent(self)
if term_info.persist_session and self._is_valid_qobject(term_info.instance):
self._park_terminal(term_info, console, avoid_console=True)
logger.info(f"Unregistered console {console_id} for terminal {terminal_id}")
return
self._delete_terminal_info(term_info)
del self._terminal_registry[terminal_id]
elif detached:
self._park_terminal(term_info, console, avoid_console=True)
logger.info(f"Unregistered console {console_id} for terminal {terminal_id}")
@@ -127,6 +234,8 @@ class BecConsoleRegistry(QWidget):
if (info := self._terminal_registry.get(console.terminal_id)) is None:
logger.warning(f"Console {console.console_id} references an unknown terminal!")
return False
if not self._is_valid_qobject(info.instance):
return False
return info.owner_console_id == console.console_id
def take_ownership(self, console: BecConsole) -> BecTerminal | None:
@@ -141,14 +250,19 @@ class BecConsoleRegistry(QWidget):
console_id, terminal_id = console.console_id, console.terminal_id
if terminal_id not in self._terminal_registry:
logger.warning(f"Terminal {terminal_id} not found in registry")
return None
self.register(console)
instance_info = self._terminal_registry[terminal_id]
if (old_owner_console_id := instance_info.owner_console_id) is not None:
if (old_owner := self._consoles.get(old_owner_console_id)) is not None:
if not self._is_valid_qobject(instance_info.instance):
self._replace_terminal(instance_info, console)
if (old_owner_console_ide := instance_info.owner_console_id) is not None:
if (
old_owner_console_ide != console_id
and (old_owner := self._consoles.get(old_owner_console_ide)) is not None
):
old_owner.yield_ownership() # call this on the old owner to make sure it is updated
instance_info.owner_console_id = console_id
instance_info.registered_console_ids.add(console_id)
logger.info(f"Transferred ownership of terminal {terminal_id} to {console_id}")
return instance_info.instance
@@ -168,17 +282,21 @@ class BecConsoleRegistry(QWidget):
return None
instance_info = self._terminal_registry[terminal_id]
if not self._is_valid_qobject(instance_info.instance):
if instance_info.owner_console_id == console_id:
self._replace_terminal(instance_info, console)
else:
return None
if instance_info.owner_console_id == console_id:
return instance_info.instance
def yield_ownership(self, console: BecConsole):
"""
Yield ownership of a instance without destroying it. The instance remains in the
Yield ownership of an instance without destroying it. The instance remains in the
registry with no owner, available for another widget to claim.
Args:
gui_id (str): The GUI ID of the widget yielding ownership.
console (BecConsole): The console which wishes to yield ownership of its associated terminal.
"""
console_id, terminal_id = console.console_id, console.terminal_id
logger.debug(f"Console {console_id} attempted to yield ownership")
@@ -190,20 +308,40 @@ class BecConsoleRegistry(QWidget):
logger.debug(f"But it was not the owner, which was {term_info.owner_console_id}!")
return
term_info.owner_console_id = None
console._detach_terminal_widget(term_info.instance)
self._park_terminal(term_info, console)
term_info.instance.setParent(QApplication.instance().console_widget_registry)
def should_initialize(self, console: BecConsole) -> bool:
"""Return true if the console should send its startup command to the terminal."""
info = self._terminal_registry.get(console.terminal_id)
if info is None:
return False
return (
info.owner_console_id == console.console_id
and not info.initialized
and self._is_valid_qobject(info.instance)
)
def mark_initialized(self, console: BecConsole) -> None:
info = self._terminal_registry.get(console.terminal_id)
if info is not None and info.owner_console_id == console.console_id:
info.initialized = True
def owner_is_visible(self, term_id: str) -> bool:
"""
Check if the owner of a instance is currently visible.
Check if the owner of an instance is currently visible.
Args:
unique_id (str): The unique identifier for the instance.
term_id (str): The terminal ID to check.
Returns:
bool: True if the owner is visible, False otherwise.
"""
instance_info = self._terminal_registry.get(term_id)
if instance_info is None or instance_info.owner_console_id is None:
if (
instance_info is None
or instance_info.owner_console_id is None
or not self._is_valid_qobject(instance_info.instance)
):
return False
if (owner := self._consoles.get(instance_info.owner_console_id)) is None:
@@ -211,6 +349,9 @@ class BecConsoleRegistry(QWidget):
return owner.isVisible()
_bec_console_registry = BecConsoleRegistry()
class _Overlay(QWidget):
def __init__(self, console: BecConsole):
super().__init__(parent=console)
@@ -227,8 +368,12 @@ class _Overlay(QWidget):
class BecConsole(BECWidget, QWidget):
"""A console widget with access to a shared registry of terminals, such that instances can be moved around."""
_js_callback = Signal(bool)
initialized = Signal()
PLUGIN = True
ICON_NAME = "terminal"
persist_terminal_session = False
def __init__(
self,
@@ -238,7 +383,6 @@ class BecConsole(BECWidget, QWidget):
gui_id=None,
startup_cmd: str | None = None,
terminal_id: str | None = None,
persevere_terminal: bool = False,
**kwargs,
):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
@@ -248,7 +392,6 @@ class BecConsole(BECWidget, QWidget):
self.terminal_id = terminal_id or str(uuid4())
self.console_id = self.gui_id
self.term: BecTerminal | None = None # Will be set in _set_up_instance
self.persevere_terminal = persevere_terminal
self._set_up_instance()
@@ -276,13 +419,12 @@ class BecConsole(BECWidget, QWidget):
self._stacked_layout.addWidget(self._overlay)
# will create a new terminal instance if there isn't already one for this ID
QApplication.instance().console_widget_registry.register(self)
_bec_console_registry.register(self)
self._infer_mode()
if self.startup_cmd:
self.write(self.startup_cmd, True) # will have no effect if not the owner
self._ensure_startup_started()
def _infer_mode(self):
self.term = QApplication.instance().console_widget_registry.try_get_term(self)
self.term = _bec_console_registry.try_get_term(self)
if self.term:
self._set_mode(ConsoleMode.ACTIVE)
elif self.isHidden():
@@ -301,8 +443,9 @@ class BecConsole(BECWidget, QWidget):
match mode:
case ConsoleMode.ACTIVE:
if self.term:
if self.term not in (self._term_layout.children()):
if self._term_layout.indexOf(self.term) == -1: # type: ignore[arg-type]
self._term_layout.addWidget(self.term) # type: ignore # BecTerminal is QWidget
self.term.show() # type: ignore[attr-defined]
self._stacked_layout.setCurrentIndex(0)
self._mode = mode
else:
@@ -326,7 +469,6 @@ class BecConsole(BECWidget, QWidget):
def startup_cmd(self, cmd: str | None):
"""
Set the startup command for the console.
logger.info(f"{self._console_id} inferred mode active through ownerp)
"""
self._startup_cmd = cmd
@@ -341,14 +483,38 @@ class BecConsole(BECWidget, QWidget):
if self.term:
self.term.write(data, send_return)
def _ensure_startup_started(self):
if not self.startup_cmd or not _bec_console_registry.should_initialize(self):
return
self.write(self.startup_cmd, True)
_bec_console_registry.mark_initialized(self)
def _detach_terminal_widget(self, term: BecTerminal | None) -> bool:
if term is None or not BecConsoleRegistry._is_valid_qobject(term):
if self.term is term:
self.term = None
return False
is_child = self.isAncestorOf(term) # type: ignore[arg-type]
if self._term_layout.indexOf(term) != -1: # type: ignore[arg-type]
self._term_layout.removeWidget(term) # type: ignore[arg-type]
is_child = True
if is_child:
term.hide() # type: ignore[attr-defined]
term.setParent(None) # type: ignore[attr-defined]
if self.term is term:
self.term = None
return is_child
def take_terminal_ownership(self):
"""
Take ownership of a web instance from the registry. This will transfer the instance
from its current owner (if any) to this widget.
"""
# Get the instance from registry
self.term = QApplication.instance().console_widget_registry.take_ownership(self)
self.term = _bec_console_registry.take_ownership(self)
self._infer_mode()
self._ensure_startup_started()
if self._mode == ConsoleMode.ACTIVE:
logger.debug(f"Widget {self.gui_id} took ownership of instance {self.terminal_id}")
@@ -358,7 +524,7 @@ class BecConsole(BECWidget, QWidget):
available for another widget to claim. This is automatically called when the
widget becomes hidden.
"""
QApplication.instance().console_widget_registry.yield_ownership(self)
_bec_console_registry.yield_ownership(self)
self._infer_mode()
if self._mode != ConsoleMode.ACTIVE:
logger.debug(f"Widget {self.gui_id} yielded ownership of instance {self.terminal_id}")
@@ -371,13 +537,13 @@ class BecConsole(BECWidget, QWidget):
def showEvent(self, event):
"""Called when the widget is shown. Updates UI state based on ownership."""
super().showEvent(event)
if not QApplication.instance().console_widget_registry.is_owner(self):
if not QApplication.instance().console_widget_registry.owner_is_visible(self.terminal_id):
if not _bec_console_registry.is_owner(self):
if not _bec_console_registry.owner_is_visible(self.terminal_id):
self.take_terminal_ownership()
def cleanup(self):
"""Unregister this console on destruction."""
QApplication.instance().console_widget_registry.unregister(self)
_bec_console_registry.unregister(self)
super().cleanup()
@@ -389,6 +555,7 @@ class BECShell(BecConsole):
"""
ICON_NAME = "hub"
persist_terminal_session = True
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
super().__init__(
@@ -397,7 +564,6 @@ class BECShell(BecConsole):
client=client,
gui_id=gui_id,
terminal_id="bec_shell",
persevere_terminal=True,
**kwargs,
)
+20 -9
View File
@@ -2,25 +2,20 @@ from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Literal
from typing import TYPE_CHECKING, Literal
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger, messages
from bec_lib.endpoints import MessageEndpoints
from bec_lib.utils.import_utils import lazy_import, lazy_import_from
from pydantic import BaseModel, Field, field_validator
from qtpy.QtCore import QObject, Qt, QThread, QTimer, Signal
from qtpy.QtGui import QTransform
from scipy.interpolate import (
CloughTocher2DInterpolator,
LinearNDInterpolator,
NearestNDInterpolator,
)
from scipy.spatial import cKDTree
from toolz import partition
from bec_widgets.utils import Colors
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbars.actions import MaterialIconAction
@@ -32,6 +27,22 @@ from bec_widgets.widgets.plots.plot_base import PlotBase
logger = bec_logger.logger
if TYPE_CHECKING:
from scipy.interpolate import (
CloughTocher2DInterpolator,
LinearNDInterpolator,
NearestNDInterpolator,
)
from scipy.spatial import cKDTree
else:
CloughTocher2DInterpolator, LinearNDInterpolator, NearestNDInterpolator = lazy_import_from(
"scipy.interpolate",
["CloughTocher2DInterpolator", "LinearNDInterpolator", "NearestNDInterpolator"],
)
cKDTree = lazy_import_from("scipy.spatial", ["cKDTree"])
class HeatmapDeviceSignal(BaseModel):
"""The configuration of a signal in the scatter waveform widget."""
@@ -611,7 +622,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"]
@@ -4,9 +4,9 @@ import os
from qtpy.QtWidgets import QFrame, QScrollArea, QVBoxLayout
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
class HeatmapSettings(SettingWidget):
+1 -1
View File
@@ -10,7 +10,7 @@ from pydantic import BaseModel, Field, field_validator
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors, apply_theme
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.widgets.plots.image.image_base import ImageBase
@@ -9,7 +9,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationError
from qtpy.QtCore import QPointF, Signal, SignalInstance
from qtpy.QtWidgets import QDialog, QVBoxLayout
from bec_widgets.utils import Colors
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.container_utils import WidgetContainerUtils
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.side_panel import SidePanel
@@ -9,7 +9,8 @@ from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Signal
from qtpy.QtGui import QTransform
from bec_widgets.utils import BECConnector, Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.widgets.plots.image.image_processor import (
ImageProcessor,
ImageStats,
@@ -20,7 +20,8 @@ from qtpy.QtWidgets import (
)
from bec_widgets import BECWidget
from bec_widgets.utils import BECDispatcher, ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.toolbars.actions import WidgetAction
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
from bec_widgets.utils.toolbars.toolbar import MaterialIconAction, ModularToolBar
@@ -10,8 +10,8 @@ from qtpy.QtCore import Signal
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QHBoxLayout, QMainWindow, QWidget
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors, apply_theme
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbars.toolbar import MaterialIconAction
@@ -2,9 +2,9 @@ import os
from qtpy.QtWidgets import QFrame, QScrollArea, QVBoxLayout, QWidget
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.utils.widget_io import WidgetIO
@@ -10,7 +10,8 @@ from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QWidget
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.side_panel import SidePanel
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
@@ -2,9 +2,9 @@ import os
from qtpy.QtWidgets import QVBoxLayout, QWidget
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.utils.widget_io import WidgetIO
+3 -1
View File
@@ -8,8 +8,10 @@ from bec_lib import bec_logger
from qtpy.QtCore import QPoint, QPointF, Qt, Signal
from qtpy.QtWidgets import QHBoxLayout, QLabel, QMainWindow, QVBoxLayout, QWidget
from bec_widgets.utils import ConnectionConfig, Crosshair, EntryValidator
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.crosshair import Crosshair
from bec_widgets.utils.entry_validator import EntryValidator
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.fps_counter import FPSCounter
from bec_widgets.utils.plot_indicator_items import BECArrowItem, BECTickItem
+1 -1
View File
@@ -10,7 +10,7 @@ from qtpy import QtCore
from qtpy.QtCore import QObject, Signal
from bec_widgets import SafeProperty
from bec_widgets.utils import BECConnector, ConnectionConfig
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import Colors
if TYPE_CHECKING:
@@ -8,7 +8,8 @@ from bec_lib import bec_logger
from pydantic import BaseModel, Field, ValidationError, field_validator
from qtpy import QtCore
from bec_widgets.utils import BECConnector, Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import Colors
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform
@@ -7,7 +7,8 @@ from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import QTimer, Signal
from qtpy.QtWidgets import QHBoxLayout, QMainWindow, QWidget
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbars.toolbar import MaterialIconAction
@@ -2,9 +2,9 @@ import os
from qtpy.QtWidgets import QFrame, QScrollArea, QVBoxLayout
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
class ScatterCurveSettings(SettingWidget):
@@ -2,9 +2,9 @@ import os
from qtpy.QtWidgets import QFrame, QScrollArea, QVBoxLayout, QWidget
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.utils.widget_io import WidgetIO
+2 -1
View File
@@ -8,7 +8,8 @@ from bec_lib import bec_logger
from pydantic import BaseModel, Field, field_validator
from qtpy import QtCore
from bec_widgets.utils import BECConnector, Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
from bec_widgets.utils.colors import Colors
if TYPE_CHECKING: # pragma: no cover
from bec_widgets.widgets.plots.waveform.waveform import Waveform
@@ -50,9 +50,10 @@ from qtpy.QtWidgets import (
)
from bec_widgets import SafeSlot
from bec_widgets.utils import ConnectionConfig, EntryValidator
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.entry_validator import EntryValidator
from bec_widgets.utils.toolbars.actions import WidgetAction
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
from bec_widgets.utils.toolbars.toolbar import MaterialIconAction, ModularToolBar
@@ -10,6 +10,7 @@ from bec_lib.device import Positioner
from bec_lib.endpoints import MessageEndpoints
from bec_lib.lmfit_serializer import serialize_lmfit_params, serialize_param_object
from bec_lib.scan_data_container import ScanDataContainer
from bec_lib.utils.import_utils import lazy_import
from pydantic import Field, ValidationError, field_validator
from qtpy.QtCore import Qt, QTimer, Signal
from qtpy.QtWidgets import (
@@ -25,7 +26,7 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.utils import ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_signal_proxy import BECSignalProxy
from bec_widgets.utils.colors import Colors, apply_theme
from bec_widgets.utils.container_utils import WidgetContainerUtils
@@ -54,13 +55,7 @@ _DAP_PARAM = object()
if TYPE_CHECKING: # pragma: no cover
import lmfit # type: ignore
else:
try:
import lmfit # type: ignore
except Exception as e: # pragma: no cover
logger.warning(
f"lmfit could not be imported: {e}. Custom DAP functionality will be unavailable."
)
lmfit = None
lmfit = lazy_import("lmfit")
# noinspection PyDataclass
@@ -6,8 +6,8 @@ from bec_lib.logger import bec_logger
from qtpy.QtCore import QPointF, QSize, Qt
from qtpy.QtWidgets import QHBoxLayout, QLabel, QVBoxLayout, QWidget
from bec_widgets.utils import Colors
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import Colors
from bec_widgets.utils.error_popups import SafeProperty
from bec_widgets.utils.settings_dialog import SettingsDialog
from bec_widgets.utils.toolbars.actions import MaterialIconAction
@@ -19,9 +19,9 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.utils import UILoader
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.settings_dialog import SettingWidget
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.progress.ring_progress_bar.ring import Ring
from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
@@ -12,10 +12,10 @@ from pyqtgraph import SignalProxy
from qtpy.QtCore import QThreadPool, Signal
from qtpy.QtWidgets import QFileDialog, QListWidget, QToolButton, QVBoxLayout, QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.services.device_browser.device_item import DeviceItem
from bec_widgets.widgets.services.device_browser.device_item.device_config_dialog import (
@@ -52,19 +52,19 @@ class BecQTerm(QWidget):
self.setLayout(self._layout)
if QTermWidget:
self._main_widget = QTermWidget(parent=self)
self.activity.connect(self._main_widget.activity)
self.bell.connect(self._main_widget.bell)
self.copy_available.connect(self._main_widget.copyAvailable)
self.current_directory_changed.connect(self._main_widget.currentDirectoryChanged)
self.finished.connect(self._main_widget.finished)
self.profile_changed.connect(self._main_widget.profileChanged)
self.received_data.connect(self._main_widget.receivedData)
self.silence.connect(self._main_widget.silence)
self.term_got_focus.connect(self._main_widget.termGetFocus)
self.term_key_pressed.connect(self._main_widget.termKeyPressed)
self.term_lost_focus.connect(self._main_widget.termLostFocus)
self.title_changed.connect(self._main_widget.titleChanged)
self.url_activated.connect(self._main_widget.urlActivated)
self._main_widget.activity.connect(self.activity)
self._main_widget.bell.connect(self.bell)
self._main_widget.copyAvailable.connect(self.copy_available)
self._main_widget.currentDirectoryChanged.connect(self.current_directory_changed)
self._main_widget.finished.connect(self.finished)
self._main_widget.profileChanged.connect(self.profile_changed)
self._main_widget.receivedData.connect(self.received_data)
self._main_widget.silence.connect(self.silence)
self._main_widget.termGetFocus.connect(self.term_got_focus)
self._main_widget.termKeyPressed.connect(self.term_key_pressed)
self._main_widget.termLostFocus.connect(self.term_lost_focus)
self._main_widget.titleChanged.connect(self.title_changed)
self._main_widget.urlActivated.connect(self.url_activated)
self._setEnvironment([f"{k}={v}" for k, v in os.environ.items()])
self._setColorScheme("Solarized")
else:
@@ -78,12 +78,6 @@ class BecQTerm(QWidget):
text += "\n"
self._sendText(text)
def deleteLater(self, /) -> None:
return super().deleteLater()
def close(self, /) -> bool:
return super().close()
# automatically forwarded to the widget only if it exists
@_forward
def _addCustomColorSchemeDir(self, custom_dir: str, /) -> None: ...
@@ -1,58 +0,0 @@
"""Utilities for filtering and formatting in the LogPanel"""
from __future__ import annotations
import re
from collections import deque
from typing import Callable, Iterator
from bec_lib.logger import LogLevel
from bec_lib.messages import LogMessage
from qtpy.QtCore import QDateTime
LinesHtmlFormatter = Callable[[deque[LogMessage]], Iterator[str]]
LineFormatter = Callable[[LogMessage], str]
LineFilter = Callable[[LogMessage], bool] | None
ANSI_ESCAPE_REGEX = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
def replace_escapes(s: str):
s = ANSI_ESCAPE_REGEX.sub("", s)
return s.replace(" ", "&nbsp;").replace("\n", "<br />").replace("\t", " ")
def level_filter(msg: LogMessage, thresh: int):
return LogLevel[msg.content["log_type"].upper()].value >= thresh
def noop_format(line: LogMessage):
_textline = line.log_msg if isinstance(line.log_msg, str) else line.log_msg["text"]
return replace_escapes(_textline.strip()) + "<br />"
def simple_color_format(line: LogMessage, colors: dict[LogLevel, str]):
color = colors.get(LogLevel[line.content["log_type"].upper()]) or colors[LogLevel.INFO]
return f'<font color="{color}">{noop_format(line)}</font>'
def create_formatter(line_format: LineFormatter, line_filter: LineFilter) -> LinesHtmlFormatter:
def _formatter(data: deque[LogMessage]):
if line_filter is not None:
return (line_format(line) for line in data if line_filter(line))
else:
return (line_format(line) for line in data)
return _formatter
def log_txt(line):
return line.log_msg if isinstance(line.log_msg, str) else line.log_msg["text"]
def log_time(line):
return QDateTime.fromMSecsSinceEpoch(int(line.log_msg["record"]["time"]["timestamp"] * 1000))
def log_svc(line):
return line.log_msg["service_name"]
@@ -30,7 +30,7 @@ class LogPanelPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return DOM_XML
def group(self):
return "BEC Services"
return ""
def icon(self):
return designer_material_icon(LogPanel.ICON_NAME)
@@ -51,7 +51,7 @@ class LogPanelPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return "LogPanel"
def toolTip(self):
return "Displays a log panel"
return "LogPanel"
def whatsThis(self):
return self.toolTip()
+432 -329
View File
@@ -2,21 +2,31 @@
from __future__ import annotations
import operator
import os
import re
from collections import deque
from functools import partial, reduce
from re import Pattern
from typing import TYPE_CHECKING, Literal
from dataclasses import dataclass
from functools import partial
from typing import Iterable, Literal
from bec_lib.client import BECClient
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import LogLevel, bec_logger
from bec_lib.messages import LogMessage, StatusMessage
from pyqtgraph import SignalProxy
from qtpy.QtCore import QDateTime, QObject, Qt, Signal
from qtpy.QtGui import QFont
from bec_qthemes import material_icon
from qtpy.QtCore import Signal # type: ignore
from qtpy.QtCore import (
QAbstractTableModel,
QCoreApplication,
QDateTime,
QModelIndex,
QObject,
QPersistentModelIndex,
QSize,
QSortFilterProxyModel,
Qt,
QTimer,
)
from qtpy.QtGui import QColor
from qtpy.QtWidgets import (
QApplication,
QCheckBox,
@@ -25,204 +35,414 @@ from qtpy.QtWidgets import (
QDialog,
QGridLayout,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QPushButton,
QScrollArea,
QTextEdit,
QSizePolicy,
QTableView,
QToolButton,
QVBoxLayout,
QWidget,
)
from thefuzz import fuzz
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.colors import apply_theme, get_theme_palette
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.editors.text_box.text_box import TextBox
from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECServiceStatusMixin
from bec_widgets.widgets.utility.logpanel._util import (
LineFilter,
LineFormatter,
LinesHtmlFormatter,
create_formatter,
level_filter,
log_svc,
log_time,
log_txt,
noop_format,
simple_color_format,
)
if TYPE_CHECKING: # pragma: no cover
from qtpy.QtCore import SignalInstance
logger = bec_logger.logger
MODULE_PATH = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
# TODO: improve log color handling
DEFAULT_LOG_COLORS = {
LogLevel.INFO: "#FFFFFF",
LogLevel.SUCCESS: "#00FF00",
LogLevel.WARNING: "#FFCC00",
LogLevel.ERROR: "#FF0000",
LogLevel.DEBUG: "#0000CC",
_DEFAULT_LOG_COLORS = {
LogLevel.INFO.name: QColor("#FFFFFF"),
LogLevel.SUCCESS.name: QColor("#00FF00"),
LogLevel.WARNING.name: QColor("#FFCC00"),
LogLevel.ERROR.name: QColor("#FF0000"),
LogLevel.DEBUG.name: QColor("#0000CC"),
}
@dataclass(frozen=True)
class _Constants:
FUZZ_THRESHOLD = 80
UPDATE_INTERVAL_MS = 200
headers = ["level", "timestamp", "service_name", "message", "function"]
_CONST = _Constants()
class TimestampUpdate:
def __init__(self, value: QDateTime | None, update_type: Literal["start", "end"]) -> None:
self.value = value
self.update_type = update_type
class BecLogsQueue(BECConnector, QObject):
"""Manages getting logs from BEC Redis and formatting them for display"""
RPC = False
new_message = Signal()
new_messages = Signal()
paused = Signal(bool)
_instance: BecLogsQueue | None = None
def __init__(
self,
parent: QObject | None,
maxlen: int = 1000,
line_formatter: LineFormatter = noop_format,
**kwargs,
) -> None:
@classmethod
def instance(cls):
if cls._instance is None:
cls._instance = cls(QCoreApplication.instance())
return cls._instance
def __init__(self, parent: QObject | None, maxlen: int = 2500, **kwargs) -> None:
if BecLogsQueue._instance:
raise RuntimeError("Create no more than one BecLogsQueue - use BecLogsQueue.instance()")
super().__init__(parent=parent, **kwargs)
self._timestamp_start: QDateTime | None = None
self._timestamp_end: QDateTime | None = None
self._max_length = maxlen
self._data: deque[LogMessage] = deque([], self._max_length)
self._display_queue: deque[str] = deque([], self._max_length)
self._log_level: str | None = None
self._search_query: Pattern | str | None = None
self._selected_services: set[str] | None = None
self._set_formatter_and_update_filter(line_formatter)
# instance attribute still accessible after c++ object is deleted, so the callback can be unregistered
self._paused = False
self._data = deque(
(
item["data"]
for item in self.bec_dispatcher.client.connector.xread(
MessageEndpoints.log(), count=self._max_length, id="0"
)
),
maxlen=self._max_length,
)
self._incoming: deque[LogMessage] = deque([], maxlen=self._max_length)
self.bec_dispatcher.connect_slot(self._process_incoming_log_msg, MessageEndpoints.log())
self._update_timer = QTimer(self, interval=_CONST.UPDATE_INTERVAL_MS)
self._update_timer.timeout.connect(self._proc_update)
QCoreApplication.instance().aboutToQuit.connect(self.cleanup) # type: ignore
self._update_timer.start()
def __len__(self):
return len(self._data)
@SafeSlot()
def toggle_pause(self):
self._paused = not self._paused
self.paused.emit(self._paused)
def row_data(self, index: int) -> LogMessage | None:
if index < 0 or index > (len(self._data) - 1):
return None
return self._data[index]
def cell_data(self, row: int, key: str):
if key == "level":
return self._data[row].log_type.upper()
msg_item = self._data[row].log_msg
if isinstance(msg_item, str):
return msg_item
if key == "service_name":
return msg_item.get(key)
elif key in ["service_name", "function", "message"]:
return msg_item.get("record", {}).get(key)
elif key == "timestamp":
return msg_item.get("record", {}).get("time", {}).get("repr")
def log_timestamp(self, row: int) -> float:
msg_item = self._data[row].log_msg
if isinstance(msg_item, str):
return 0
return msg_item.get("record", {}).get("time", {}).get("timestamp")
def cleanup(self, *_):
"""Stop listening to the Redis log stream"""
self.bec_dispatcher.disconnect_slot(
self._process_incoming_log_msg, [MessageEndpoints.log()]
)
self._update_timer.stop()
BecLogsQueue._instance = None
@SafeSlot(verify_sender=True)
def _process_incoming_log_msg(self, msg: dict, _metadata: dict):
try:
_msg = LogMessage(**msg)
self._data.append(_msg)
if self.filter is None or self.filter(_msg):
self._display_queue.append(self._line_formatter(_msg))
self.new_message.emit()
self._incoming.append(_msg)
except Exception as e:
if "Internal C++ object (BecLogsQueue) already deleted." in e.args:
return
logger.warning(f"Error in LogPanel incoming message callback: {e}")
def _set_formatter_and_update_filter(self, line_formatter: LineFormatter = noop_format):
self._line_formatter: LineFormatter = line_formatter
self._queue_formatter: LinesHtmlFormatter = create_formatter(
self._line_formatter, self.filter
)
def _combine_filters(self, *args: LineFilter):
return lambda msg: reduce(operator.and_, [filt(msg) for filt in args if filt is not None])
def _create_re_filter(self) -> LineFilter:
if self._search_query is None:
return None
elif isinstance(self._search_query, str):
return lambda line: self._search_query in log_txt(line)
return lambda line: self._search_query.match(log_txt(line)) is not None
def _create_service_filter(self):
return (
lambda line: self._selected_services is None or log_svc(line) in self._selected_services
)
def _create_timestamp_filter(self) -> LineFilter:
s, e = self._timestamp_start, self._timestamp_end
if s is e is None:
return lambda msg: True
def _time_filter(msg):
msg_time = log_time(msg)
if s is None:
return msg_time <= e
if e is None:
return s <= msg_time
return s <= msg_time <= e
return _time_filter
@property
def filter(self) -> LineFilter:
"""A function which filters a log message based on all applied criteria"""
thresh = LogLevel[self._log_level].value if self._log_level is not None else 0
return self._combine_filters(
partial(level_filter, thresh=thresh),
self._create_re_filter(),
self._create_timestamp_filter(),
self._create_service_filter(),
)
def update_level_filter(self, level: str):
"""Change the log-level of the level filter"""
if level not in [l.name for l in LogLevel]:
logger.error(f"Logging level {level} unrecognized for filter!")
@SafeSlot(verify_sender=True)
def _proc_update(self):
if self._paused or len(self._incoming) == 0:
return
self._log_level = level
self._set_formatter_and_update_filter(self._line_formatter)
self._data.extend(self._incoming)
self._incoming.clear()
self.new_messages.emit()
def update_search_filter(self, search_query: Pattern | str | None = None):
"""Change the string or regex to filter against"""
self._search_query = search_query
self._set_formatter_and_update_filter(self._line_formatter)
def update_time_filter(self, start: QDateTime | None, end: QDateTime | None):
"""Change the start and/or end times to filter against"""
self._timestamp_start = start
self._timestamp_end = end
self._set_formatter_and_update_filter(self._line_formatter)
class BecLogsTableModel(QAbstractTableModel):
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.log_queue = BecLogsQueue.instance()
self.log_queue.new_messages.connect(self.handle_new_messages)
self._headers = _CONST.headers
def update_service_filter(self, services: set[str]):
"""Change the selected services to display"""
self._selected_services = services
self._set_formatter_and_update_filter(self._line_formatter)
def rowCount(self, parent: QModelIndex | QPersistentModelIndex = QModelIndex()) -> int:
return len(self.log_queue)
def update_line_formatter(self, line_formatter: LineFormatter):
"""Update the formatter"""
self._set_formatter_and_update_filter(line_formatter)
def columnCount(self, parent: QModelIndex | QPersistentModelIndex = QModelIndex()) -> int:
return len(self._headers)
def display_all(self) -> str:
"""Return formatted output for all log messages"""
return "\n".join(self._queue_formatter(self._data.copy()))
def headerData(self, section, orientation, role=int(Qt.ItemDataRole.DisplayRole)):
if role == Qt.ItemDataRole.DisplayRole and orientation == Qt.Orientation.Horizontal:
return self._headers[section]
return None
def format_new(self):
"""Return formatted output for the display queue"""
res = "\n".join(self._display_queue)
self._display_queue = deque([], self._max_length)
return res
def get_row_data(self, index: QModelIndex) -> LogMessage | None:
"""Return the row data for the given index."""
if not index.isValid():
return None
return self.log_queue.row_data(index.row())
def clear_logs(self):
"""Clear the cache and display queue"""
self._data = deque([])
self._display_queue = deque([])
def timestamp(self, row: int):
return QDateTime.fromMSecsSinceEpoch(int(self.log_queue.log_timestamp(row) * 1000))
def fetch_history(self):
"""Fetch all available messages from Redis"""
self._data = deque(
item["data"]
for item in self.bec_dispatcher.client.connector.xread(
MessageEndpoints.log().endpoint, from_start=True, count=self._max_length
)
def data(self, index, role=int(Qt.ItemDataRole.DisplayRole)):
"""Return data for the given index and role."""
if not index.isValid():
return
if role in [Qt.ItemDataRole.DisplayRole, Qt.ItemDataRole.ToolTipRole]:
return self.log_queue.cell_data(index.row(), self._headers[index.column()])
if role in [Qt.ItemDataRole.ForegroundRole]:
return self._map_log_level_color(self.log_queue.cell_data(index.row(), "level"))
def _map_log_level_color(self, data):
return _DEFAULT_LOG_COLORS.get(data)
def handle_new_messages(self):
self.dataChanged.emit(
self.index(0, 0), self.index(self.rowCount() - 1, self.columnCount() - 1)
)
def unique_service_names_from_history(self) -> set[str]:
"""Go through the log history to determine active service names"""
return set(msg.log_msg["service_name"] for msg in self._data)
class LogMsgProxyModel(QSortFilterProxyModel):
show_service_column = Signal(bool)
def __init__(
self,
parent=None,
service_filter: set[str] | None = None,
level_filter: LogLevel | None = None,
):
super().__init__(parent)
self._service_filter = service_filter or set()
self._level_filter: LogLevel | None = level_filter
self._filter_text: str = ""
self._fuzzy_search: bool = False
self._time_filter_start: QDateTime | None = None
self._time_filter_end: QDateTime | None = None
def get_row_data(self, rows: Iterable[QModelIndex]) -> Iterable[LogMessage | None]:
return (self.sourceModel().get_row_data(self.mapToSource(idx)) for idx in rows)
def sourceModel(self) -> BecLogsTableModel:
return super().sourceModel() # type: ignore
@SafeSlot(int, int)
def refresh(self, *_):
self.invalidateRowsFilter()
@SafeSlot(None)
@SafeSlot(set)
def update_service_filter(self, filter: set[str]):
"""Filter to the selected services (show any service in the provided set)
Args:
filter (set[str] | None): set of services for which to show logs"""
self._service_filter = filter
self.show_service_column.emit(len(filter) != 1)
self.invalidateRowsFilter()
@SafeSlot(None)
@SafeSlot(LogLevel)
def update_level_filter(self, filter: LogLevel | None):
"""Filter to the selected log level
Args:
filter (str | None): lowest log level to show"""
self._level_filter = filter
self.invalidateRowsFilter()
@SafeSlot(str)
def update_filter_text(self, filter: str):
"""Filter messages based on text
Args:
filter (str | None): set of services for which to show logs"""
self._filter_text = filter
self.invalidateRowsFilter()
@SafeSlot(bool)
def update_fuzzy(self, state: bool):
"""Set text filter to fuzzy search or not
Args:
state (bool): fuzzy search on"""
self._fuzzy_search = state
self.invalidateRowsFilter()
@SafeSlot(TimestampUpdate)
def update_timestamp(self, update: TimestampUpdate):
if update.update_type == "start":
self._time_filter_start = update.value
else:
self._time_filter_end = update.value
self.invalidateRowsFilter()
def filterAcceptsRow(self, source_row: int, source_parent) -> bool:
# No service filter, and no filter text, display everything
possible_filters = [
self._service_filter,
self._level_filter,
self._filter_text,
self._time_filter_start,
self._time_filter_end,
]
if not any(map(bool, possible_filters)):
return True
model = self.sourceModel()
# Filter out services
if self._service_filter:
col = _CONST.headers.index("service_name")
if model.data(model.index(source_row, col, source_parent)) not in self._service_filter:
return False
# Filter out levels
if self._level_filter:
col = _CONST.headers.index("level")
level: str = model.data(model.index(source_row, col, source_parent)) # type: ignore
if LogLevel[level] < self._level_filter:
return False
# Filter time
if self._time_filter_start:
if model.timestamp(source_row) < self._time_filter_start:
return False
if self._time_filter_end:
if model.timestamp(source_row) > self._time_filter_end:
return False
# Filter message text - must go last because this can return True
if self._filter_text:
col = _CONST.headers.index("message")
msg: str = model.data(model.index(source_row, col, source_parent)).lower() # type: ignore
if self._fuzzy_search:
return fuzz.partial_ratio(self._filter_text.lower(), msg) >= _CONST.FUZZ_THRESHOLD
else:
return self._filter_text.lower() in msg.lower()
return True
class BecLogTableView(QTableView):
def __init__(self, *args, max_message_width: int = 1000, **kwargs) -> None:
super().__init__(*args, **kwargs)
header = QHeaderView(Qt.Orientation.Horizontal, parent=self)
header.setSectionResizeMode(QHeaderView.ResizeMode.Interactive)
header.setStretchLastSection(True)
header.setMaximumSectionSize(max_message_width)
self.setHorizontalHeader(header)
def model(self) -> LogMsgProxyModel:
return super().model() # type: ignore
class LogPanel(BECWidget, QWidget):
"""Live display of the BEC logs in a table view."""
PLUGIN = True
ICON_NAME = "browse_activity"
def __init__(
self,
parent: QWidget | None = None,
max_message_width: int = 1000,
show_toolbar: bool = True,
service_filter: set[str] | None = None,
level_filter: LogLevel | None = None,
**kwargs,
) -> None:
super().__init__(parent=parent, **kwargs)
self._setup_models(service_filter=service_filter, level_filter=level_filter)
self._layout = QVBoxLayout()
self.setLayout(self._layout)
if show_toolbar:
self._setup_toolbar(client=self.client)
self._setup_table_view(max_message_width=max_message_width)
self._update_service_filter(service_filter or set())
if show_toolbar:
self._connect_toolbar()
self._proxy.show_service_column.connect(self._show_service_column)
colors = QApplication.instance().theme.accent_colors # type: ignore
dict_colors = QApplication.instance().theme.colors # type: ignore
_DEFAULT_LOG_COLORS.update(
{
LogLevel.INFO.name: dict_colors["FG"],
LogLevel.SUCCESS.name: colors.success,
LogLevel.WARNING.name: colors.warning,
LogLevel.ERROR.name: colors.emergency,
LogLevel.DEBUG.name: dict_colors["BORDER"],
}
)
self._table.scrollToBottom()
def _setup_models(self, service_filter: set[str] | None, level_filter: LogLevel | None):
self._model = BecLogsTableModel(parent=self)
self._proxy = LogMsgProxyModel(
parent=self, service_filter=service_filter, level_filter=level_filter
)
self._proxy.setSourceModel(self._model)
self._model.log_queue.new_messages.connect(self._proxy.refresh)
def _setup_table_view(self, max_message_width: int) -> None:
"""Setup the table view."""
self._table = BecLogTableView(self, max_message_width=max_message_width)
self._table.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
self._layout.addWidget(self._table)
self._table.setModel(self._proxy)
self._table.setHorizontalScrollMode(QTableView.ScrollMode.ScrollPerPixel)
self._table.setTextElideMode(Qt.TextElideMode.ElideRight)
self._table.resizeColumnsToContents()
def _setup_toolbar(self, client: BECClient):
self._toolbar = LogPanelToolbar(self, client)
self._layout.addWidget(self._toolbar)
def _connect_toolbar(self):
self._toolbar.services_selected.connect(self._proxy.update_service_filter)
self._toolbar.search_textbox.textChanged.connect(self._proxy.update_filter_text)
self._toolbar.level_changed.connect(self._proxy.update_level_filter)
self._toolbar.fuzzy_changed.connect(self._proxy.update_fuzzy)
self._toolbar.timestamp_update.connect(self._proxy.update_timestamp)
self._toolbar.pause_button.clicked.connect(self._model.log_queue.toggle_pause)
self._model.log_queue.paused.connect(self._toolbar._update_pause_button_icon)
def _update_service_filter(self, filter: set[str]):
self._service_filter = filter
self._proxy.update_service_filter(filter)
self._table.setColumnHidden(
_CONST.headers.index("service_name"), len(self._service_filter) == 1
)
@SafeSlot(bool)
def _show_service_column(self, show: bool):
self._table.setColumnHidden(_CONST.headers.index("service_name"), not show)
def sizeHint(self) -> QSize:
return QSize(600, 300)
class LogPanelToolbar(QWidget):
services_selected = Signal(set)
level_changed = Signal(LogLevel)
fuzzy_changed = Signal(bool)
timestamp_update = Signal(TimestampUpdate)
services_selected: SignalInstance = Signal(set)
def __init__(self, parent: QWidget | None = None) -> None:
def __init__(self, parent: QWidget | None = None, client: BECClient | None = None) -> None:
"""A toolbar for the logpanel, mainly used for managing the states of filters"""
super().__init__(parent)
@@ -231,51 +451,69 @@ class LogPanelToolbar(QWidget):
self._timestamp_end: QDateTime | None = None
self._unique_service_names: set[str] = set()
self._services_selected: set[str] | None = None
self._services_selected: set[str] = set()
self.layout = QHBoxLayout(self) # type: ignore
self._layout = QHBoxLayout(self)
self.service_choice_button = QPushButton("Select services", self)
self.layout.addWidget(self.service_choice_button)
self.service_choice_button.clicked.connect(self._open_service_filter_dialog)
if client is not None:
self.client = client
self.service_choice_button = QPushButton("Select services", self)
self._layout.addWidget(self.service_choice_button)
self.service_choice_button.clicked.connect(self._open_service_filter_dialog)
self.service_list_update(self.client.service_status)
self._services_selected = self._unique_service_names
self.filter_level_dropdown = self._log_level_box()
self.layout.addWidget(self.filter_level_dropdown)
self.clear_button = QPushButton("Clear all", self)
self.layout.addWidget(self.clear_button)
self.fetch_button = QPushButton("Fetch history", self)
self.layout.addWidget(self.fetch_button)
self._layout.addWidget(self.filter_level_dropdown)
self.filter_level_dropdown.currentTextChanged.connect(self._emit_level)
self._string_search_box()
self.timerange_button = QPushButton("Set time range", self)
self.layout.addWidget(self.timerange_button)
self._layout.addWidget(self.timerange_button)
self.timerange_button.clicked.connect(self._open_datetime_dialog)
@property
def time_start(self):
return self._timestamp_start
self.pause_button = QToolButton()
self.pause_button.setIcon(material_icon("pause", size=(20, 20), convert_to_pixmap=False))
self._PLAYING_TOOLTIP = "Pause live log updates."
self._PAUSED_TOOLTIP = "Continue live log updates."
self.pause_button.setToolTip(self._PLAYING_TOOLTIP)
self._layout.addWidget(self.pause_button)
@property
def time_end(self):
return self._timestamp_end
@SafeSlot(bool)
def _update_pause_button_icon(self, paused):
if paused:
icon = "play_arrow"
tooltip = self._PAUSED_TOOLTIP
else:
icon = "pause"
tooltip = self._PLAYING_TOOLTIP
self.pause_button.setIcon(material_icon(icon, size=(20, 20), convert_to_pixmap=False))
self.pause_button.setToolTip(tooltip)
def _string_search_box(self):
self.layout.addWidget(QLabel("Search: "))
self._layout.addWidget(QLabel("Search: "))
self.search_textbox = QLineEdit()
self.layout.addWidget(self.search_textbox)
self.layout.addWidget(QLabel("Use regex: "))
self.regex_enabled = QCheckBox()
self.layout.addWidget(self.regex_enabled)
self.update_re_button = QPushButton("Update search", self)
self.layout.addWidget(self.update_re_button)
self._layout.addWidget(self.search_textbox)
self._layout.addWidget(QLabel("Fuzzy: "))
self.fuzzy = QCheckBox()
self._layout.addWidget(self.fuzzy)
self.fuzzy.checkStateChanged.connect(self._emit_fuzzy)
def _log_level_box(self):
box = QComboBox()
box.setToolTip("Display logs with equal or greater significance to the selected level.")
[box.addItem(l.name) for l in LogLevel]
[box.addItem(level.name) for level in LogLevel]
return box
@SafeSlot(str)
def _emit_level(self, level: str):
self.level_changed.emit(LogLevel[level])
@SafeSlot(Qt.CheckState)
def _emit_fuzzy(self, state: Qt.CheckState):
self.fuzzy_changed.emit(state == Qt.CheckState.Checked)
def _current_ts(self, selection_type: Literal["start", "end"]):
if selection_type == "start":
return self._timestamp_start
@@ -284,6 +522,7 @@ class LogPanelToolbar(QWidget):
else:
raise ValueError(f"timestamps can only be for the start or end, not {selection_type}")
@SafeSlot()
def _open_datetime_dialog(self):
"""Open dialog window for timestamp filter selection"""
self._dt_dialog = QDialog(self)
@@ -312,8 +551,8 @@ class LogPanelToolbar(QWidget):
)
_layout.addWidget(date_clear_button)
for v in [("start", label_start), ("end", label_end)]:
date_button_set(*v)
date_button_set("start", label_start)
date_button_set("end", label_end)
close_button = QPushButton("Close", parent=self._dt_dialog)
close_button.clicked.connect(self._dt_dialog.accept)
@@ -352,27 +591,23 @@ class LogPanelToolbar(QWidget):
self._timestamp_start = dt
else:
self._timestamp_end = dt
self.timestamp_update.emit(TimestampUpdate(value=dt, update_type=selection_type))
@SafeSlot(dict, set)
def service_list_update(
self, services_info: dict[str, StatusMessage], services_from_history: set[str], *_, **__
):
def service_list_update(self, services_info: dict[str, StatusMessage]):
"""Change the list of services which can be selected"""
self._unique_service_names = set([s.split("/")[0] for s in services_info.keys()])
self._unique_service_names |= services_from_history
if self._services_selected is None:
self._services_selected = self._unique_service_names
@SafeSlot()
def _open_service_filter_dialog(self):
self.service_list_update(self.client.service_status)
if len(self._unique_service_names) == 0 or self._services_selected is None:
return
self._svc_dialog = QDialog(self)
self._svc_dialog.setWindowTitle(f"Select services to show logs from")
self._svc_dialog.setWindowTitle("Select services to show logs from")
layout = QVBoxLayout()
self._svc_dialog.setLayout(layout)
service_cb_grid = QGridLayout(parent=self._svc_dialog)
service_cb_grid = QGridLayout()
layout.addLayout(service_cb_grid)
def check_box(name: str, checked: Qt.CheckState):
@@ -398,146 +633,6 @@ class LogPanelToolbar(QWidget):
self._svc_dialog.deleteLater()
class LogPanel(TextBox):
"""Displays a log panel"""
ICON_NAME = "terminal"
service_list_update = Signal(dict, set)
def __init__(
self,
parent=None,
client: BECClient | None = None,
service_status: BECServiceStatusMixin | None = None,
**kwargs,
):
"""Initialize the LogPanel widget."""
super().__init__(parent=parent, client=client, config={"text": ""}, **kwargs)
self._update_colors()
self._service_status = service_status or BECServiceStatusMixin(self, client=self.client) # type: ignore
self._log_manager = BecLogsQueue(
parent=self, line_formatter=partial(simple_color_format, colors=self._colors)
)
self._proxy_update = SignalProxy(
self._log_manager.new_message, rateLimit=1, slot=self._on_append
)
self.toolbar = LogPanelToolbar(parent=self)
self.toolbar_area = QScrollArea()
self.toolbar_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
self.toolbar_area.setSizeAdjustPolicy(QScrollArea.SizeAdjustPolicy.AdjustToContents)
self.toolbar_area.setFixedHeight(int(self.toolbar.clear_button.height() * 2))
self.toolbar_area.setWidget(self.toolbar)
self.layout.addWidget(self.toolbar_area)
self.toolbar.clear_button.clicked.connect(self._on_clear)
self.toolbar.fetch_button.clicked.connect(self._on_fetch)
self.toolbar.update_re_button.clicked.connect(self._on_re_update)
self.toolbar.search_textbox.returnPressed.connect(self._on_re_update)
self.toolbar.regex_enabled.checkStateChanged.connect(self._on_re_update)
self.toolbar.filter_level_dropdown.currentTextChanged.connect(self._set_level_filter)
self.toolbar.timerange_button.clicked.connect(self._choose_datetime)
self._service_status.services_update.connect(self._update_service_list)
self.service_list_update.connect(self.toolbar.service_list_update)
self.toolbar.services_selected.connect(self._update_service_filter)
self.text_box_text_edit.setFont(QFont("monospace", 12))
self.text_box_text_edit.setHtml("")
self.text_box_text_edit.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap)
self._connect_to_theme_change()
@SafeSlot(set)
def _update_service_filter(self, services: set[str]):
self._log_manager.update_service_filter(services)
self._on_redraw()
@SafeSlot(dict, dict)
def _update_service_list(self, services_info: dict[str, StatusMessage], *_, **__):
self.service_list_update.emit(
services_info, self._log_manager.unique_service_names_from_history()
)
@SafeSlot()
def _choose_datetime(self):
self.toolbar._open_datetime_dialog()
self._set_time_filter()
def _connect_to_theme_change(self):
"""Connect to the theme change signal."""
qapp = QApplication.instance()
if hasattr(qapp, "theme_signal"):
qapp.theme_signal.theme_updated.connect(self._on_redraw) # type: ignore
def _update_colors(self):
self._colors = DEFAULT_LOG_COLORS.copy()
self._colors.update({LogLevel.INFO: get_theme_palette().text().color().name()})
def _cursor_to_end(self):
c = self.text_box_text_edit.textCursor()
c.movePosition(c.MoveOperation.End)
self.text_box_text_edit.setTextCursor(c)
@SafeSlot()
@SafeSlot(str)
def _on_redraw(self, *_):
self._update_colors()
self._log_manager.update_line_formatter(partial(simple_color_format, colors=self._colors))
self.set_html_text(self._log_manager.display_all())
self._cursor_to_end()
@SafeSlot(verify_sender=True)
def _on_append(self, *_):
self.text_box_text_edit.insertHtml(self._log_manager.format_new())
self._cursor_to_end()
@SafeSlot()
def _on_clear(self):
self._log_manager.clear_logs()
self.set_html_text(self._log_manager.display_all())
self._cursor_to_end()
@SafeSlot()
@SafeSlot(Qt.CheckState)
def _on_re_update(self, *_):
if self.toolbar.regex_enabled.isChecked():
try:
search_query = re.compile(self.toolbar.search_textbox.text())
except Exception as e:
logger.warning(f"Failed to compile search regex with error {e}")
search_query = None
logger.info(f"Setting LogPanel search regex to {search_query}")
else:
search_query = self.toolbar.search_textbox.text()
logger.info(f'Setting LogPanel search string to "{search_query}"')
self._log_manager.update_search_filter(search_query)
self.set_html_text(self._log_manager.display_all())
self._cursor_to_end()
@SafeSlot()
def _on_fetch(self):
self._log_manager.fetch_history()
self.set_html_text(self._log_manager.display_all())
self._cursor_to_end()
@SafeSlot(str)
def _set_level_filter(self, level: str):
self._log_manager.update_level_filter(level)
self._on_redraw()
@SafeSlot()
def _set_time_filter(self):
self._log_manager.update_time_filter(self.toolbar.time_start, self.toolbar.time_end)
self._on_redraw()
def cleanup(self):
self._service_status.cleanup()
self._log_manager.cleanup()
self._log_manager.deleteLater()
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
@@ -545,7 +640,15 @@ if __name__ == "__main__": # pragma: no cover
app = QApplication(sys.argv)
apply_theme("dark")
widget = LogPanel()
panel = QWidget()
queue = BecLogsQueue(panel)
layout = QVBoxLayout(panel)
layout.addWidget(QLabel("All logs, no filters:"))
layout.addWidget(LogPanel())
layout.addWidget(QLabel("All services, level filter WARNING preapplied:"))
layout.addWidget(LogPanel(level_filter=LogLevel.WARNING))
layout.addWidget(QLabel('All services, service filter {"DeviceServer"} preapplied:'))
layout.addWidget(LogPanel(service_filter={"DeviceServer"}))
widget.show()
panel.show()
sys.exit(app.exec())
@@ -3,8 +3,8 @@ from qtpy import QtCore, QtGui
from qtpy.QtCore import Property, Signal, Slot
from qtpy.QtWidgets import QSizePolicy, QVBoxLayout, QWidget
from bec_widgets.utils import Colors
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import Colors
class RoundedColorMapButton(ColorMapButton):
@@ -19,7 +19,7 @@ from qtpy.QtWidgets import (
QWidget,
)
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.widget_highlighter import WidgetHighlighter
from bec_widgets.utils.widget_io import WidgetHierarchy
+10 -11
View File
@@ -1,6 +1,6 @@
[project]
name = "bec_widgets"
version = "3.4.2"
version = "3.8.0"
description = "BEC Widgets"
requires-python = ">=3.11"
classifiers = [
@@ -12,19 +12,19 @@ dependencies = [
"PyJWT~=2.9",
"PySide6==6.9.0",
"PySide6-QtAds==4.4.0",
"bec_ipython_client~=3.107,>=3.107.2", # needed for jupyter console
"bec_ipython_client~=3.107,>=3.107.2", # needed for jupyter console
"bec_lib~=3.107,>=3.107.2",
"bec_qthemes~=1.0, >=1.3.4",
"black>=26,<27", # needed for bw-generate-cli
"black>=26,<27", # needed for bw-generate-cli
"copier~=9.7",
"darkdetect~=0.8",
"isort>=5.13, <9.0", # needed for bw-generate-cli
"isort>=5.13, <9.0", # needed for bw-generate-cli
"markdown~=3.9",
"ophyd_devices~=1.29, >=1.29.1",
"pydantic~=2.0",
"pylsp-bec~=1.2",
"pyqtgraph==0.13.7",
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtconsole~=5.5, >=5.5.1", # needed for jupyter console
"qtmonaco~=0.8, >=0.8.1",
"qtpy~=2.4",
"thefuzz~=0.22",
@@ -38,8 +38,8 @@ Homepage = "https://gitlab.psi.ch/bec/bec_widgets"
[project.scripts]
bec-app = "bec_widgets.applications.main_app:main"
bec-designer = "bec_widgets.utils.bec_designer:main"
bec-gui-server = "bec_widgets.cli.server:main"
bw-generate-cli = "bec_widgets.cli.generate_cli:main"
bec-gui-server = "bec_widgets.applications.companion_app:main"
bw-generate-cli = "bec_widgets.utils.generate_cli:main"
[project.optional-dependencies]
dev = [
@@ -52,12 +52,11 @@ dev = [
"pytest-xvfb~=3.0",
"pytest~=8.0",
"pytest-cov~=6.1.1",
"pytest-benchmark~=5.2",
"watchdog~=6.0",
"pre_commit~=4.2",
]
qtermwidget = [
"pyside6_qtermwidget",
]
qtermwidget = ["pyside6_qtermwidget"]
[build-system]
requires = ["hatchling"]
@@ -68,7 +67,7 @@ line-length = 100
skip-magic-trailing-comma = true
[tool.coverage.report]
skip_empty = true # exclude empty *files*, e.g. __init__.py, from the report
skip_empty = true # exclude empty *files*, e.g. __init__.py, from the report
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
@@ -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()
+18 -5
View File
@@ -1,3 +1,5 @@
import traceback
import pytest
import qtpy.QtCore
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
@@ -5,12 +7,14 @@ from qtpy.QtCore import QTimer
class TestableQTimer(QTimer):
_instances: list[tuple[QTimer, str]] = []
_instances: list[tuple[QTimer, str, str]] = []
_current_test_name: str = ""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
TestableQTimer._instances.append((self, TestableQTimer._current_test_name))
tb = traceback.format_stack()
init_line = list(filter(lambda msg: "QTimer(" in msg, tb))[-1]
TestableQTimer._instances.append((self, TestableQTimer._current_test_name, init_line))
@classmethod
def check_all_stopped(cls, qtbot):
@@ -20,12 +24,21 @@ class TestableQTimer(QTimer):
except RuntimeError as e:
return "already deleted" in e.args[0]
def _format_timers(timers: list[tuple[QTimer, str, str]]):
return "\n".join(
f"Timer: {t[0]}\n in test: {t[1]}\n created at:{t[2]}" for t in timers
)
try:
qtbot.waitUntil(lambda: all(_is_done_or_deleted(timer) for timer, _ in cls._instances))
qtbot.waitUntil(
lambda: all(_is_done_or_deleted(timer) for timer, _, _ in cls._instances)
)
except QtBotTimeoutError as exc:
active_timers = list(filter(lambda t: t[0].isActive(), cls._instances))
(t.stop() for t, _ in cls._instances)
raise TimeoutError(f"Failed to stop all timers: {active_timers}") from exc
(t.stop() for t, _, _ in cls._instances)
raise TimeoutError(
f"Failed to stop all timers:\n{_format_timers(active_timers)}"
) from exc
cls._instances = []
+4 -3
View File
@@ -33,7 +33,7 @@ def threads_check_fixture(threads_check):
@pytest.fixture
def gui_id():
"""New gui id each time, to ensure no 'gui is alive' zombie key can perturb"""
return f"figure_{random.randint(0,100)}" # make a new gui id each time, to ensure no 'gui is alive' zombie key can perturb
return f"figure_{random.randint(0, 100)}" # make a new gui id each time, to ensure no 'gui is alive' zombie key can perturb
@pytest.fixture(scope="function")
@@ -51,6 +51,7 @@ def connected_client_gui_obj(qtbot, gui_id, bec_client_lib):
qtbot.waitUntil(lambda: len(gui.bec.widget_list()) == 0, timeout=10000)
yield gui
finally:
gui.bec.delete_all() # ensure clean state
qtbot.waitUntil(lambda: len(gui.bec.widget_list()) == 0, timeout=10000)
if (bec := getattr(gui, "bec", None)) is not None:
bec.delete_all() # ensure clean state
qtbot.waitUntil(lambda: len(bec.widget_list()) == 0, timeout=10000)
gui.kill_server()
+2 -1
View File
@@ -1,6 +1,7 @@
import pytest
from bec_widgets.cli.client import Image, MotorMap, Waveform
from bec_widgets.cli.client_utils import BECGuiClient
from bec_widgets.cli.rpc.rpc_base import RPCReference
# pylint: disable=unused-argument
@@ -122,7 +123,7 @@ def test_ring_bar(qtbot, connected_client_gui_obj):
assert gui._ipython_registry[bar._gui_id].__class__.__name__ == "RingProgressBar"
def test_rpc_gui_obj(connected_client_gui_obj, qtbot):
def test_rpc_gui_obj(connected_client_gui_obj: BECGuiClient, qtbot):
gui = connected_client_gui_obj
qtbot.waitUntil(lambda: len(gui.windows) == 1, timeout=3000)
-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:
@@ -260,22 +260,6 @@ def test_widgets_e2e_image(qtbot, connected_client_gui_obj, random_generator_fro
maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
# TODO re-enable when issue is resolved #560
# @pytest.mark.timeout(PYTEST_TIMEOUT)
# def test_widgets_e2e_log_panel(qtbot, connected_client_gui_obj, random_generator_from_seed):
# """Test the LogPanel widget."""
# gui = connected_client_gui_obj
# bec = gui._client
# # Create dock_area and widget
# widget = create_widget(qtbot, gui, gui.available_widgets.LogPanel)
# widget: client.LogPanel
# # No rpc calls to check so far
# # Test removing the widget, or leaving it open for the next test
# maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
@pytest.mark.timeout(PYTEST_TIMEOUT)
def test_widgets_e2e_minesweeper(qtbot, connected_client_gui_obj, random_generator_from_seed):
"""Test the MineSweeper widget."""
@@ -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
+10 -5
View File
@@ -1,3 +1,8 @@
# Force ophyd onto its dummy control layer in tests so importing it does not
# try to create a real EPICS CA context.
import os
os.environ.setdefault("OPHYD_CONTROL_LAYER", "dummy")
import json
import time
from unittest import mock
@@ -13,16 +18,16 @@ from bec_lib.client import BECClient
from bec_lib.messages import _StoredDataInfo
from bec_qthemes import apply_theme
from bec_qthemes._theme import Theme
from ophyd._pyepics_shim import _dispatcher
from ophyd._dummy_shim import _dispatcher
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
from qtpy.QtCore import QEvent, QEventLoop
from qtpy.QtWidgets import QApplication, QMessageBox
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.tests.utils import DEVICES, DMMock
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
from bec_widgets.utils import error_popups
from bec_widgets.utils.bec_dispatcher import QtRedisConnector
from bec_widgets.utils.rpc_register import RPCRegister
# Patch to set default RAISE_ERROR_DEFAULT to True for tests
# This means that by default, error popups will raise exceptions during tests
@@ -222,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"],
@@ -269,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": {
@@ -349,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 -1
View File
@@ -5,7 +5,7 @@ import pytest
from qtpy.QtCore import QObject
from qtpy.QtWidgets import QApplication, QWidget
from bec_widgets.utils import BECConnector
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.error_popups import SafeProperty
from bec_widgets.utils.error_popups import SafeSlot as Slot
+119 -6
View File
@@ -1,10 +1,13 @@
from unittest import mock
import pytest
from qtpy.QtCore import Qt
import shiboken6
from qtpy.QtCore import QEvent, QEventLoop, Qt
from qtpy.QtGui import QHideEvent, QShowEvent
from qtpy.QtTest import QTest
from qtpy.QtWidgets import QApplication, QWidget
import bec_widgets.widgets.editors.bec_console.bec_console as bec_console_module
from bec_widgets.widgets.editors.bec_console.bec_console import (
BecConsole,
BECShell,
@@ -15,6 +18,20 @@ from bec_widgets.widgets.editors.bec_console.bec_console import (
from .client_mocks import mocked_client
def process_deferred_deletes():
app = QApplication.instance()
app.sendPostedEvents(None, QEvent.DeferredDelete)
app.processEvents(QEventLoop.AllEvents)
@pytest.fixture(autouse=True)
def clean_bec_console_registry():
_bec_console_registry.clear()
yield
_bec_console_registry.clear()
process_deferred_deletes()
@pytest.fixture
def console_widget(qtbot):
"""Create a BecConsole widget."""
@@ -128,12 +145,108 @@ def test_is_owner(console_widget: BecConsole):
assert not _bec_console_registry.is_owner(mock_console)
def test_bec_shell_leaves_terminal_instantiated(qtbot):
widget = BECShell()
def test_closing_active_console_keeps_terminal_valid_for_remaining_console(qtbot):
widget1 = BecConsole(client=mocked_client, gui_id="close_owner", terminal_id="shared_close")
widget2 = BecConsole(client=mocked_client, gui_id="remaining", terminal_id="shared_close")
qtbot.addWidget(widget2)
widget1.take_terminal_ownership()
term = widget1.term
assert term is not None
widget1.close()
widget1.deleteLater()
process_deferred_deletes()
assert shiboken6.isValid(term)
widget2.take_terminal_ownership()
assert widget2.term is term
assert widget2._mode == ConsoleMode.ACTIVE
def test_active_console_detaches_terminal_before_destruction(qtbot):
widget1 = BecConsole(client=mocked_client, gui_id="owner", terminal_id="shared_detach")
widget2 = BecConsole(client=mocked_client, gui_id="survivor", terminal_id="shared_detach")
qtbot.addWidget(widget1)
qtbot.addWidget(widget2)
widget1.take_terminal_ownership()
term = widget1.term
assert term is not None
assert widget1.isAncestorOf(term)
widget1.close()
assert shiboken6.isValid(term)
assert not widget1.isAncestorOf(term)
assert term.parent() is widget2._term_holder
def test_bec_shell_terminal_persists_after_last_shell_unregisters(qtbot):
shell = BECShell(gui_id="bec_shell_persistent")
qtbot.addWidget(shell)
term = shell.term
assert term is not None
_bec_console_registry.unregister(shell)
info = _bec_console_registry._terminal_registry["bec_shell"]
assert info.registered_console_ids == set()
assert info.owner_console_id is None
assert info.persist_session is True
assert info.instance is term
assert shiboken6.isValid(term)
def test_new_bec_shell_claims_preserved_terminal(qtbot):
shell1 = BECShell(gui_id="bec_shell_first")
term = shell1.term
assert term is not None
shell1.close()
shell1.deleteLater()
process_deferred_deletes()
assert "bec_shell" in _bec_console_registry._terminal_registry
assert shiboken6.isValid(term)
shell2 = BECShell(gui_id="bec_shell_second")
qtbot.addWidget(shell2)
shell2.showEvent(QShowEvent())
assert shell2.term is term
assert shell2._mode == ConsoleMode.ACTIVE
def test_persistent_bec_shell_sends_startup_command_once(qtbot, monkeypatch):
class RecordingTerminal(QWidget):
writes = []
def write(self, text: str, add_newline: bool = True):
self.writes.append((text, add_newline))
monkeypatch.setattr(bec_console_module, "_BecTermClass", RecordingTerminal)
shell1 = BECShell(gui_id="bec_shell_startup_first")
shell1.close()
shell1.deleteLater()
process_deferred_deletes()
shell2 = BECShell(gui_id="bec_shell_startup_second")
qtbot.addWidget(shell2)
shell2.showEvent(QShowEvent())
assert len(RecordingTerminal.writes) == 1
assert RecordingTerminal.writes[0][0].startswith("bec ")
assert RecordingTerminal.writes[0][1] is True
def test_plain_console_terminal_removed_after_last_unregister(qtbot):
widget = BecConsole(client=mocked_client, gui_id="plain_console", terminal_id="plain_terminal")
qtbot.addWidget(widget)
assert len(_bec_console_registry._terminal_registry) != 0
assert "plain_terminal" in _bec_console_registry._terminal_registry
_bec_console_registry.unregister(widget)
assert len(_bec_console_registry._terminal_registry) != 0
assert _bec_console_registry._terminal_registry["bec_shell"].owner_console_id is None
assert "plain_terminal" not in _bec_console_registry._terminal_registry
-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,
}
],
@@ -9,7 +9,8 @@ from bec_widgets.cli.rpc.rpc_base import RPCBase
from bec_widgets.utils.plugin_utils import BECClassContainer, BECClassInfo
class _TestGlobalPlugin(RPCBase): ...
class _TestGlobalPlugin(RPCBase):
_IMPORT_MODULE = "test.global.plugin.widgets"
mock_client_module_globals = SimpleNamespace()
@@ -25,12 +26,13 @@ mock_client_module_globals.Widgets = _TestGlobalPlugin
def test_plugins_dont_clobber_client_globals(bec_logger: MagicMock):
reload(client)
bec_logger.logger.warning.assert_called_with(
"Plugin widget Widgets from namespace(Widgets=<class 'tests.unit_tests.test_client_plugin_widgets._TestGlobalPlugin'>) conflicts with a built-in class!"
"Plugin widget Widgets in test.global.plugin.widgets conflicts with a built-in class!"
)
assert isinstance(client.Widgets, enum.EnumType)
class _TestDuplicatePlugin(RPCBase): ...
class _TestDuplicatePlugin(RPCBase):
_IMPORT_MODULE = "test.duplicate.plugin.module"
mock_client_module_duplicate = SimpleNamespace()
@@ -54,7 +56,7 @@ def test_duplicate_plugins_not_allowed(_, bec_logger: MagicMock):
reload(client)
assert (
call(
f"Detected duplicate widget Waveform in plugin repo file: {inspect.getfile(_TestDuplicatePlugin)} !"
"Plugin widget Waveform in test.duplicate.plugin.module conflicts with a built-in class!"
)
in bec_logger.logger.warning.mock_calls
)
+2 -2
View File
@@ -4,9 +4,9 @@ from pydantic import ValidationError
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QVBoxLayout, QWidget
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.colors import Colors, apply_theme
from bec_widgets.widgets.plots.waveform.curve import CurveConfig
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget
+1 -1
View File
@@ -4,7 +4,7 @@ import pytest
from qtpy.QtCore import QPointF, Qt
from qtpy.QtGui import QTransform
from bec_widgets.utils import Crosshair
from bec_widgets.utils.crosshair import Crosshair
from bec_widgets.widgets.plots.image.image_item import ImageItem
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from tests.unit_tests.client_mocks import mocked_client
+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
+333 -207
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)
@@ -239,7 +239,7 @@ class TestBasicDockArea:
assert basic_dock_area.widget_map(bec_widgets_only=False)["panel_bec"] is panel_bec
def test_new_widget_string_creates_widget(self, basic_dock_area, qtbot):
basic_dock_area.new("DarkModeButton")
basic_dock_area.new("RingProgressBar")
qtbot.waitUntil(lambda: len(basic_dock_area.dock_list()) > 0, timeout=1000)
assert basic_dock_area.widget_list()
@@ -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."""
@@ -652,7 +623,7 @@ class TestDockManagement:
initial_count = len(advanced_dock_area.dock_list())
# Create a widget by string name
widget = advanced_dock_area.new("DarkModeButton")
widget = advanced_dock_area.new("RingProgressBar")
# Wait for the dock to be created (since it's async)
qtbot.wait(200)
@@ -720,7 +691,7 @@ class TestDockManagement:
initial_count = len(widget_map)
# Create a widget
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(200)
# Check widget map updated
@@ -734,7 +705,7 @@ class TestDockManagement:
initial_count = len(widget_list)
# Create a widget
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(200)
# Check widget list updated
@@ -744,8 +715,8 @@ class TestDockManagement:
def test_delete_all(self, advanced_dock_area, qtbot):
"""Test delete_all functionality."""
# Create multiple widgets
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
advanced_dock_area.new("RingProgressBar")
# Wait for docks to be created
qtbot.wait(200)
@@ -801,7 +772,7 @@ class TestWorkspaceLocking:
def test_lock_workspace_property_setter(self, advanced_dock_area, qtbot):
"""Test workspace_is_locked property setter."""
# Create a dock first
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(200)
# Initially unlocked
@@ -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 = [
@@ -926,8 +887,8 @@ class TestToolbarFunctionality:
def test_attach_all_action(self, advanced_dock_area, qtbot):
"""Test attach_all toolbar action."""
# Create floating docks
advanced_dock_area.new("DarkModeButton", start_floating=True)
advanced_dock_area.new("DarkModeButton", start_floating=True)
advanced_dock_area.new("RingProgressBar", start_floating=True)
advanced_dock_area.new("RingProgressBar", start_floating=True)
qtbot.wait(200)
@@ -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")
@@ -955,7 +916,7 @@ class TestToolbarFunctionality:
# Floating entry
settings.setArrayIndex(0)
settings.setValue("object_name", "FloatingWaveform")
settings.setValue("widget_class", "DarkModeButton")
settings.setValue("widget_class", "RingProgressBar")
settings.setValue("closable", True)
settings.setValue("floatable", True)
settings.setValue("movable", True)
@@ -973,7 +934,7 @@ class TestToolbarFunctionality:
# Anchored entry
settings.setArrayIndex(1)
settings.setValue("object_name", "EmbeddedWaveform")
settings.setValue("widget_class", "DarkModeButton")
settings.setValue("widget_class", "RingProgressBar")
settings.setValue("closable", True)
settings.setValue("floatable", True)
settings.setValue("movable", True)
@@ -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,29 +1740,29 @@ 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")
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
advanced_dock_area.new("RingProgressBar")
advanced_dock_area.new("RingProgressBar")
# Wait for docks to be created
qtbot.wait(1000)
@@ -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,11 +1866,11 @@ 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")
settings.setValue("widget_class", "DarkModeButton")
settings.setValue("widget_class", "RingProgressBar")
settings.setValue("closable", True)
settings.setValue("floatable", True)
settings.setValue("movable", True)
@@ -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,11 +1972,11 @@ 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")
settings.setValue("widget_class", "DarkModeButton")
settings.setValue("widget_class", "RingProgressBar")
settings.setValue("closable", True)
settings.setValue("floatable", True)
settings.setValue("movable", True)
@@ -1844,7 +1985,7 @@ class TestWorkspaceProfileOperations:
advanced_dock_area.load_profile(source_profile)
qtbot.wait(500)
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(500)
class StubDialog:
@@ -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,11 +2025,11 @@ 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")
settings.setValue("widget_class", "DarkModeButton")
settings.setValue("widget_class", "RingProgressBar")
settings.setValue("closable", True)
settings.setValue("floatable", True)
settings.setValue("movable", True)
@@ -1892,13 +2038,13 @@ class TestWorkspaceProfileOperations:
advanced_dock_area.load_profile(profile_a)
qtbot.wait(500)
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(500)
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 (
@@ -2229,7 +2361,6 @@ class TestFlatToolbarActions:
"flat_progress_bar",
"flat_terminal",
"flat_bec_shell",
"flat_log_panel",
"flat_sbb_monitor",
]
@@ -2289,11 +2420,6 @@ class TestFlatToolbarActions:
action.trigger()
mock_new.assert_called_once_with(widget_type)
def test_flat_log_panel_action_disabled(self, advanced_dock_area):
"""Test that flat log panel action is disabled."""
action = advanced_dock_area.toolbar.components.get_action("flat_log_panel").action
assert not action.isEnabled()
class TestModeTransitions:
"""Test mode transitions and state consistency."""
@@ -2342,8 +2468,8 @@ class TestModeTransitions:
def test_mode_switching_preserves_existing_docks(self, advanced_dock_area, qtbot):
"""Test that mode switching doesn't affect existing docked widgets."""
# Create some widgets
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("DarkModeButton")
advanced_dock_area.new("RingProgressBar")
advanced_dock_area.new("RingProgressBar")
qtbot.wait(200)
initial_dock_count = len(advanced_dock_area.dock_list())
+35 -16
View File
@@ -5,7 +5,8 @@ import black
import isort
import pytest
from bec_widgets.cli.generate_cli import ClientGenerator
from bec_widgets.utils.generate_cli import ClientGenerator, write_designer_plugins
from bec_widgets.utils.generate_designer_plugin import DesignerPluginInfo
from bec_widgets.utils.plugin_utils import BECClassContainer, BECClassInfo
# pylint: disable=missing-function-docstring
@@ -59,6 +60,14 @@ class MockViewWithContent:
"""Activate view."""
class MockDesignerWidgetBase:
ICON_NAME = "mock_icon"
class MockDesignerWidget(MockDesignerWidgetBase):
pass
def test_client_generator_with_black_formatting():
generator = ClientGenerator(base=True)
container = BECClassContainer()
@@ -104,8 +113,7 @@ def test_client_generator_with_black_formatting():
from bec_lib.logger import bec_logger
from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call, rpc_timeout
from bec_widgets.utils.bec_plugin_helper import (get_all_plugin_widgets,
get_plugin_client_module)
from bec_widgets.utils.bec_plugin_helper import get_plugin_client_module
logger = bec_logger.logger
@@ -123,31 +131,25 @@ def test_client_generator_with_black_formatting():
try:
_plugin_widgets = get_all_plugin_widgets().as_dict()
plugin_client = get_plugin_client_module()
Widgets = _WidgetsEnumType("Widgets", {name: name for name in _plugin_widgets} | _Widgets)
if (_overlap := _Widgets.keys() & _plugin_widgets.keys()) != set():
for _widget in _overlap:
logger.warning(f"Detected duplicate widget {_widget} in plugin repo file: {inspect.getfile(_plugin_widgets[_widget])} !")
for plugin_name, plugin_class in inspect.getmembers(plugin_client, inspect.isclass):
if issubclass(plugin_class, RPCBase) and plugin_class is not RPCBase:
if plugin_name not in _Widgets:
_Widgets[plugin_name] = plugin_name
if plugin_name in globals():
conflicting_file = (
inspect.getfile(_plugin_widgets[plugin_name])
if plugin_name in _plugin_widgets
else f"{plugin_client}"
)
logger.warning(
f"Plugin widget {plugin_name} from {conflicting_file} conflicts with a built-in class!"
f"Plugin widget {plugin_name} in {plugin_class._IMPORT_MODULE} conflicts with a built-in class!"
)
continue
if plugin_name not in _overlap:
else:
globals()[plugin_name] = plugin_class
Widgets = _WidgetsEnumType("Widgets", _Widgets)
except ImportError as e:
logger.error(f"Failed loading plugins: \\n{reduce(add, traceback.format_exception(e))}")
class MockBECFigure(RPCBase):
_IMPORT_MODULE = "tests.unit_tests.test_generate_cli_client"
@rpc_call
def add_plot(self, plot_id: str):
"""
@@ -162,6 +164,8 @@ def test_client_generator_with_black_formatting():
class MockBECWaveform1D(RPCBase):
_IMPORT_MODULE = "tests.unit_tests.test_generate_cli_client"
@rpc_call
def set_frequency(self, frequency: float) -> list:
"""
@@ -290,3 +294,18 @@ c = a + b"""
content = file.read()
assert corrected in content
def test_write_designer_plugins(tmp_path):
file_name = tmp_path / "designer_plugins.py"
write_designer_plugins([DesignerPluginInfo(MockDesignerWidget)], str(file_name))
with open(file_name, "r", encoding="utf-8") as file:
content = file.read()
assert '"MockDesignerWidget":' in content
assert '"tests.unit_tests.test_generate_cli_client"' in content
assert '"MockDesignerWidget"' in content
assert '"MockDesignerWidget": "mock_icon"' in content
assert "MockDesignerWidgetPlugin" not in content
+106 -146
View File
@@ -7,163 +7,123 @@ from collections import deque
from unittest.mock import MagicMock, patch
import pytest
from bec_lib.logger import LogLevel
from bec_lib.messages import LogMessage
from bec_lib.redis_connector import StreamMessage
from qtpy.QtCore import QDateTime
from bec_widgets.widgets.utility.logpanel._util import (
log_time,
replace_escapes,
simple_color_format,
)
from bec_widgets.widgets.utility.logpanel.logpanel import DEFAULT_LOG_COLORS, LogPanel
from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel, TimestampUpdate
from .client_mocks import mocked_client
TEST_TABLE_STRING = "2025-01-15 15:57:18 | bec_server.scan_server.scan_queue | [INFO] | \n \x1b[3m primary queue / ScanQueueStatus.RUNNING \x1b[0m\n┏━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┓\n\x1b[1m \x1b[0m\x1b[1m queue_id \x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_id\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mis_scan\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mtype\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_numb…\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mIQ status\x1b[0m\x1b[1m \x1b[0m┃\n┡━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━┩\n│ bbe50c82-6… │ None │ False │ mv │ None │ PENDING │\n└─────────────┴─────────┴─────────┴──────┴────────────┴───────────┘\n\n"
TEST_LOG_MESSAGES = [
LogMessage(
metadata={},
log_type="debug",
log_msg={
"text": "datetime | debug | test log message",
"record": {"time": {"timestamp": 123456789.000}},
"service_name": "ScanServer",
},
),
LogMessage(
metadata={},
log_type="info",
log_msg={
"text": "datetime | info | test log message",
"record": {"time": {"timestamp": 123456789.007}},
"service_name": "ScanServer",
},
),
LogMessage(
metadata={},
log_type="success",
log_msg={
"text": "datetime | success | test log message",
"record": {"time": {"timestamp": 123456789.012}},
"service_name": "ScanServer",
},
),
]
TEST_COMBINED_PLAINTEXT = "datetime | debug | test log message\ndatetime | info | test log message\ndatetime | success | test log message\n"
@pytest.fixture
def raw_queue():
yield deque(TEST_LOG_MESSAGES, maxlen=100)
@pytest.fixture
def log_panel(qtbot, mocked_client: MagicMock):
widget = LogPanel(client=mocked_client, service_status=MagicMock())
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_log_panel_init(log_panel: LogPanel):
assert log_panel.plain_text == ""
def test_table_string_processing():
assert "\x1b" in TEST_TABLE_STRING
sanitized = replace_escapes(TEST_TABLE_STRING)
assert "\x1b" not in sanitized
assert " " not in sanitized
assert "\n" not in sanitized
@pytest.mark.parametrize(
["msg", "color"], zip(TEST_LOG_MESSAGES, ["#0000CC", "#FFFFFF", "#00FF00"])
)
def test_color_format(msg: LogMessage, color: str):
assert color in simple_color_format(msg, DEFAULT_LOG_COLORS)
def test_logpanel_output(qtbot, log_panel: LogPanel):
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
log_panel._on_redraw()
assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT
def display_queue_empty():
print(log_panel._log_manager._display_queue)
return len(log_panel._log_manager._display_queue) == 0
next_text = "datetime | error | test log message"
msg = LogMessage(
metadata={},
log_type="error",
log_msg={"text": next_text, "record": {}, "service_name": "ScanServer"},
)
log_panel._log_manager._process_incoming_log_msg(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
qtbot.waitUntil(display_queue_empty, timeout=5000)
assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT + next_text + "\n"
def test_level_filter(log_panel: LogPanel):
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
log_panel._log_manager.update_level_filter("INFO")
log_panel._on_redraw()
assert (
log_panel.plain_text
== "datetime | info | test log message\ndatetime | success | test log message\n"
)
def test_clear_button(log_panel: LogPanel):
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
log_panel.toolbar.clear_button.click()
assert log_panel._log_manager._data == deque([])
def test_timestamp_filter(log_panel: LogPanel):
log_panel._log_manager._timestamp_start = QDateTime(1973, 11, 29, 21, 33, 9, 5, 1)
pytest.approx(log_panel._log_manager._timestamp_start.toMSecsSinceEpoch() / 1000, 123456789.005)
log_panel._log_manager._timestamp_end = QDateTime(1973, 11, 29, 21, 33, 9, 10, 1)
pytest.approx(log_panel._log_manager._timestamp_end.toMSecsSinceEpoch() / 1000, 123456789.010)
filter_ = log_panel._log_manager._create_timestamp_filter()
assert not filter_(TEST_LOG_MESSAGES[0])
assert filter_(TEST_LOG_MESSAGES[1])
assert not filter_(TEST_LOG_MESSAGES[2])
def test_error_handling_in_callback(log_panel: LogPanel):
log_panel._log_manager.new_message = MagicMock()
with patch("bec_widgets.widgets.utility.logpanel.logpanel.logger") as logger:
# generally errors should be logged
log_panel._log_manager.new_message.emit = MagicMock(
side_effect=ValueError("Something went wrong")
)
msg = LogMessage(
{"data": msg}
for msg in [
LogMessage(
metadata={},
log_type="debug",
log_msg={
"text": "datetime | debug | test log message",
"record": {"time": {"timestamp": 123456789.000}},
"record": {
"time": {"timestamp": 123456789.000, "repr": "2025-01-01 00:00:01"},
"message": "test debug message abcd",
"function": "_debug",
},
"service_name": "ScanServer",
},
),
LogMessage(
metadata={},
log_type="info",
log_msg={
"text": "datetime | info | test info log message",
"record": {
"time": {"timestamp": 123456789.007, "repr": "2025-01-01 00:00:02"},
"message": "test info message efgh",
"function": "_info",
},
"service_name": "DeviceServer",
},
),
LogMessage(
metadata={},
log_type="success",
log_msg={
"text": "datetime | success | test log message",
"record": {
"time": {"timestamp": 123456789.012, "repr": "2025-01-01 00:00:03"},
"message": "test success message ijkl",
"function": "_success",
},
"service_name": "ScanServer",
},
),
]
]
@pytest.fixture
def log_panel(qtbot, mocked_client):
mocked_client.connector.xread = lambda *_, **__: TEST_LOG_MESSAGES
widget = LogPanel()
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
widget._model.log_queue.cleanup()
widget.close()
widget.deleteLater()
qtbot.wait(100)
def test_log_panel_init(qtbot, log_panel: LogPanel):
assert log_panel
def test_log_panel_filters(qtbot, log_panel: LogPanel):
assert log_panel._proxy.rowCount() == 3
# Service filter
log_panel._update_service_filter({"DeviceServer"})
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
log_panel._update_service_filter(set())
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
# Text filter
log_panel._proxy.update_filter_text("efgh")
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
log_panel._proxy.update_filter_text("")
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
# Time filter
log_panel._proxy.update_timestamp(
TimestampUpdate(value=QDateTime.fromMSecsSinceEpoch(123456789004), update_type="start")
)
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 2, timeout=200)
log_panel._proxy.update_timestamp(
TimestampUpdate(value=QDateTime.fromMSecsSinceEpoch(123456789009), update_type="end")
)
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
log_panel._proxy.update_timestamp(TimestampUpdate(value=None, update_type="start"))
log_panel._proxy.update_timestamp(TimestampUpdate(value=None, update_type="end"))
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
# Level filter
log_panel._proxy.update_level_filter(LogLevel.SUCCESS)
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
log_panel._proxy.update_level_filter(None)
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
def test_log_panel_update(qtbot, log_panel: LogPanel):
log_panel._model.log_queue._incoming.append(
LogMessage(
metadata={},
log_type="error",
log_msg={
"text": "datetime | error | test log message",
"record": {
"time": {"timestamp": 123456789.015, "repr": "2025-01-01 00:00:03"},
"message": "test error message xyz",
"function": "_error",
},
"service_name": "ScanServer",
},
)
log_panel._log_manager._process_incoming_log_msg(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
logger.warning.assert_called_once()
# this specific error should be ignored and not relogged
log_panel._log_manager.new_message.emit = MagicMock(
side_effect=RuntimeError("Internal C++ object (BecLogsQueue) already deleted.")
)
log_panel._log_manager._process_incoming_log_msg(
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
)
logger.warning.assert_called_once()
)
log_panel._model.log_queue._proc_update()
qtbot.waitUntil(lambda: log_panel._model.rowCount() == 4, timeout=500)

Some files were not shown because too many files have changed in this diff Show More