"""Small versioned workspace persistence and validation.""" import base64 import json from pathlib import Path from tempfile import NamedTemporaryFile from typing import Any VERSION = 6 SYSTEM_PANELS = [ {"id": "dashboard", "title": "Dashboard", "type": "dashboard", "config": {}}, {"id": "data", "title": "Data / playback", "type": "data", "config": {}}, {"id": "signals", "title": "Signal browser", "type": "signal_browser", "config": {}}, {"id": "derived", "title": "Derived units", "type": "derive_manager", "config": {}}, ] VISUALIZATION_TYPES = {"numeric", "gauge", "graph", "numeric_group", "gauge_group"} SYSTEM_TYPES = {panel["type"] for panel in SYSTEM_PANELS} SUPPORTED_PANEL_TYPES = SYSTEM_TYPES | VISUALIZATION_TYPES _SYSTEM_TYPES_BY_ID = {panel["id"]: panel["type"] for panel in SYSTEM_PANELS} _RESERVED_TYPES_BY_ID = {**_SYSTEM_TYPES_BY_ID, "numeric": "numeric", "gauge": "gauge", "graph": "graph"} DEFAULT_PANELS = [ *SYSTEM_PANELS, {"id": "numeric", "title": "Numeric readout", "type": "numeric", "config": {"signal_id": None, "decimals": 2}}, {"id": "gauge", "title": "Gauge", "type": "gauge", "config": {"signal_id": None, "decimals": 2}}, { "id": "graph", "title": "Time graph", "type": "graph", "config": {"signal_ids": [], "follow_latest": True, "zero_baseline": False, "right_zero_baseline": False}, }, ] class WorkspaceError(ValueError): pass def new_workspace() -> dict[str, Any]: return {"version": VERSION, "panels": []} def _validate_panels(panels: list[Any]) -> None: panel_ids: set[str] = set() for panel in panels: if ( not isinstance(panel, dict) or not all(isinstance(panel.get(key), str) and panel[key] for key in ("id", "title", "type")) or not isinstance(panel.get("config", {}), dict) or ("visible" in panel and not isinstance(panel["visible"], bool)) ): raise WorkspaceError("Workspace contains an invalid panel definition.") panel_id = panel["id"] panel_type = panel["type"] if panel_id in panel_ids: raise WorkspaceError("Workspace contains duplicate panel IDs.") panel_ids.add(panel_id) if panel_id in _RESERVED_TYPES_BY_ID: if panel_type != _RESERVED_TYPES_BY_ID[panel_id]: raise WorkspaceError(f"Reserved panel ID {panel_id!r} has the wrong type.") elif panel_type not in VISUALIZATION_TYPES: raise WorkspaceError(f"Unsupported panel type {panel_type!r}.") config = panel.get("config", {}) if panel_type in {"numeric", "gauge"} and config.get("signal_id") is not None and not isinstance( config["signal_id"], str ): raise WorkspaceError("Panel signal selections must be IDs or null.") if panel_type in {"numeric", "gauge"} and ( type(config.get("decimals", 2)) is not int or not 0 <= config.get("decimals", 2) <= 8 ): raise WorkspaceError("Display decimal precision must be between 0 and 8.") if panel_type in {"numeric_group", "gauge_group"}: kind = panel_type.removesuffix("_group") items = config.get("items", []) if config.get("kind", kind) != kind or not isinstance(items, list) or any( not isinstance(item, dict) or (item.get("signal_id") is not None and not isinstance(item["signal_id"], str)) or type(item.get("decimals", 2)) is not int or not 0 <= item.get("decimals", 2) <= 8 for item in items ): raise WorkspaceError("Instrument group settings are invalid.") if panel_type == "derive_manager": units = config.get("units", []) if not isinstance(units, list) or any( not isinstance(unit, dict) or not isinstance(unit.get("path"), str) or (unit.get("unit_id") is not None and not isinstance(unit["unit_id"], str)) or not isinstance(unit.get("binding_intent", False), bool) for unit in units ): raise WorkspaceError("Derived-unit library settings are invalid.") if panel_type == "graph" and ( not isinstance(config.get("signal_ids", []), list) or not all(isinstance(signal_id, str) for signal_id in config.get("signal_ids", [])) or len(config.get("signal_ids", [])) != len(set(config.get("signal_ids", []))) or not isinstance(config.get("follow_latest", True), bool) or not isinstance(config.get("zero_baseline", False), bool) or not isinstance(config.get("right_zero_baseline", False), bool) ): raise WorkspaceError("Graph panel settings are invalid.") def _validate_grid(layout: Any, panels: list[dict[str, Any]]) -> None: def valid_rect(rect: Any) -> bool: return ( isinstance(rect, list) and len(rect) == 4 and all(type(value) is int and 0 <= value <= 10000 for value in rect) and rect[2] > 0 and rect[3] > 0 ) if not isinstance(layout, dict) or layout.get("version") != 1 or not isinstance(layout.get("panels"), dict): raise WorkspaceError("Workspace grid layout is invalid.") panel_types = {panel["id"]: panel["type"] for panel in panels} if set(layout["panels"]) != set(panel_types): raise WorkspaceError("Workspace grid does not match its panel list.") for panel_id, item in layout["panels"].items(): if not isinstance(item, dict) or item.get("type") != panel_types[panel_id] or not valid_rect(item.get("rect")): raise WorkspaceError("Workspace grid panel placement is invalid.") groups = layout.get("groups", []) if not isinstance(groups, list): raise WorkspaceError("Workspace tab groups are invalid.") grouped: set[str] = set() group_ids: set[str] = set() for group in groups: if not isinstance(group, dict): raise WorkspaceError("Workspace tab group is invalid.") members = group.get("panels") group_id = group.get("id") if ( not isinstance(group_id, str) or not group_id or group_id in group_ids or not isinstance(members, list) or len(members) < 2 or any(not isinstance(member, str) or member not in panel_types or member in grouped for member in members) or len(set(members)) != len(members) or not valid_rect(group.get("rect")) or (group.get("active") is not None and group["active"] not in members) ): raise WorkspaceError("Workspace tab group is invalid.") grouped.update(members) group_ids.add(group_id) floating = layout.get("floating", {}) if not isinstance(floating, dict) or any( item not in panel_types or item in grouped for item in floating ): raise WorkspaceError("Workspace floating panels are invalid.") if any( not isinstance(rect, list) or len(rect) != 4 or any(type(value) is not int for value in rect) or rect[2] <= 0 or rect[3] <= 0 for rect in floating.values() ): raise WorkspaceError("Workspace floating geometry is invalid.") hidden = layout.get("hidden", []) if not isinstance(hidden, list) or any(not isinstance(item, str) or item not in panel_types for item in hidden): raise WorkspaceError("Workspace hidden panels are invalid.") def validate_workspace(data: dict[str, Any]) -> None: """Validate a current workspace before the window mutates its dock state.""" if not isinstance(data, dict) or data.get("version") != VERSION: raise WorkspaceError(f"Unsupported workspace format (expected version {VERSION}).") panels = data.get("panels") if not isinstance(panels, list): raise WorkspaceError("Workspace must contain a panel list.") _validate_panels(panels) if "grid_layout" in data: _validate_grid(data["grid_layout"], panels) for key in ("geometry",): if key in data and data[key] is not None and not isinstance(data[key], str): raise WorkspaceError(f"Workspace field {key!r} must be encoded text.") if data.get(key): try: base64.b64decode(data[key], validate=True) except ValueError as error: raise WorkspaceError(f"Workspace field {key!r} is invalid.") from error def load_workspace(path: str | Path) -> dict[str, Any]: try: data = json.loads(Path(path).read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise WorkspaceError(f"Could not open workspace: {error}") from error validate_workspace(data) return data def save_workspace(path: str | Path, data: dict[str, Any]) -> None: validate_workspace(data) target = Path(path) text = json.dumps(data, indent=2) + "\n" temporary: Path | None = None try: with NamedTemporaryFile( mode="w", encoding="utf-8", dir=target.parent, prefix=f".{target.name}.", suffix=".tmp", delete=False, ) as stream: temporary = Path(stream.name) stream.write(text) temporary.replace(target) finally: if temporary is not None: temporary.unlink(missing_ok=True)