Coverage for collectra/gui/backend.py: 84.68%
966 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 09:56 +0000
1"""
2API backend for annotation display logic via pywebview's js_api interface.
4JavaScript usage:
5 await pywebview.api.load_yaml("path/to/results.yaml")
6 const result = await pywebview.api.get_display_value("node_id")
7 const nodes = await pywebview.api.get_all_nodes()
8"""
10import base64
11import enum
12import platform
13import re
14import threading
15from pathlib import Path
16from typing import TYPE_CHECKING
18import webview
20from collectra.commons.files import CollectraFile
22from .data_display import CollectraAnnotationNode, CollectraGraph, NodeDisplayValue
24if TYPE_CHECKING:
25 from collectra.pipelines.base import Collectra
28class ImageFormat(enum.Enum):
29 PNG = "image/png"
30 JPEG = "image/jpeg"
31 JPG = "image/jpg"
32 GIF = "image/gif"
33 BMP = "image/bmp"
34 WEBP = "image/webp"
35 TIFF = "image/tiff"
36 TIF = "image/tiff"
39class GUIBackend:
40 """
41 API class exposed to JavaScript via pywebview.
43 All public methods are accessible in JS as:
44 window.pywebview.api.methodName(args)
45 """
47 def __init__(self, pipeline: "Collectra", inputs=()):
48 """
49 Initialize the API instance.
51 Sets up the annotation graph, YAML path, and window reference.
52 All values are initially None until load_yaml() is called.
53 """
54 self._git_lock = threading.Lock()
55 self._text_save_lock = threading.Lock()
56 self._update_lock = threading.Lock()
57 self._restart_requested = False
58 self._restart_required = False
59 self._pipeline = pipeline
60 self._collectra_file: CollectraFile | None = None
61 self._graph: CollectraGraph | None = None
62 self._yaml_path: str | None = None
63 self._window = None
64 self._collectra_folders: list[dict] = []
65 self._parent_folder: str | None = None
66 self._global_labels: set[str] = set()
67 self._global_label_counts: dict[str, int] = {} # {label: total_count}
68 from .session_cache import FolderSessionCache
70 pipeline_path = getattr(pipeline, "path", None)
71 self._session_cache = (
72 FolderSessionCache(pipeline_path) if pipeline_path else None
73 )
74 from .arrangement import ArrangementStore
76 metadata = getattr(pipeline, "pipeline_metadata", {})
77 layouts = metadata.get("layouts")
78 default_layout = None
79 if pipeline_path and isinstance(layouts, list) and layouts:
80 if isinstance(layouts[0], str) and layouts[0].strip():
81 default_layout = Path(pipeline_path) / layouts[0]
82 self._arrangement = (
83 ArrangementStore(
84 pipeline_path,
85 pipeline_name=metadata.get("name") or getattr(pipeline, "name", None),
86 default_layout=default_layout,
87 )
88 if pipeline_path
89 else None
90 )
91 self._initial_items = (
92 self._load_input_paths(inputs) if inputs else self._restore_folder_session()
93 )
95 def _save_folder_session(self, folders=None) -> None:
96 if self._session_cache is not None:
97 paths = (
98 folders
99 if folders is not None
100 else [folder["path"] for folder in self._collectra_folders]
101 )
102 self._session_cache.write([str(Path(path).resolve()) for path in paths])
104 def _restore_folder_session(self):
105 if self._session_cache is None:
106 return None
107 self._collectra_folders = []
108 seen = set()
109 for value in self._session_cache.read():
110 try:
111 path = Path(value).resolve()
112 if path in seen or not path.is_dir():
113 continue
114 scan = self._scan_collectra_folder(path)
115 if scan is None:
116 continue
117 seen.add(path)
118 self._collectra_folders.append(
119 {"name": path.name, "path": str(path), **scan}
120 )
121 except (OSError, ValueError):
122 continue
123 self._save_folder_session()
124 if not self._collectra_folders:
125 return None
126 self._parent_folder = str(Path(self._collectra_folders[0]["path"]).parent)
127 active = self._session_cache.read_active()
128 active_index = next(
129 (
130 index
131 for index, folder in enumerate(self._collectra_folders)
132 if folder["path"] == active
133 ),
134 0,
135 )
136 return {
137 **self._folder_list_result(),
138 "provided": True,
139 "mode": "parent",
140 "active_index": active_index,
141 }
143 def git_action(self, action: str, message: str = "", repository: str = "") -> dict:
144 """Operate on the Git repository containing the active results file."""
145 from .source_control import run_git
147 if not self._yaml_path:
148 return {"success": False, "error": "Open a results folder first"}
149 if not self._git_lock.acquire(blocking=False):
150 return {"success": False, "error": "A Git operation is already running"}
151 try:
152 return run_git(Path(self._yaml_path).parent, action, message, repository)
153 finally:
154 self._git_lock.release()
156 def get_arrangement(self) -> dict:
157 """The saved panel/tab arrangement for this pipeline, or None."""
158 if self._arrangement is None:
159 return {"success": True, "arrangement": None}
160 return {"success": True, "arrangement": self._arrangement.read()}
162 def save_arrangement(self, arrangement: dict) -> dict:
163 """Remember the panel/tab arrangement (per user, per pipeline)."""
164 from .arrangement import sanitize_arrangement
166 clean = sanitize_arrangement(arrangement)
167 if clean is None:
168 return {"success": False, "error": "Not a valid arrangement"}
169 if self._arrangement is not None:
170 self._arrangement.write(clean)
171 return {"success": True}
173 def _available_layouts(self):
174 metadata = getattr(self._pipeline, "pipeline_metadata", {})
175 entries = []
176 layouts = metadata.get("layouts", [])
177 for index, filename in enumerate(layouts if isinstance(layouts, list) else []):
178 if isinstance(filename, str) and filename.strip():
179 entries.append((f"pipeline:{index}", "pipeline", Path(filename).stem,
180 Path(self._pipeline.path) / filename))
181 if self._arrangement:
182 for path in self._arrangement.custom_layouts():
183 entries.append((f"custom:{path.name}", "custom", path.stem, path))
184 return entries
186 def restart_gui(self) -> dict:
187 """Close the GUI so start() can replace the process with fresh code."""
188 if self._window is None:
189 return {"success": False, "error": "No GUI window is open."}
190 if not self._update_lock.acquire(blocking=False):
191 return {"success": False, "error": "Wait for the update to finish before restarting."}
192 try:
193 if self._collectra_folders:
194 self._save_folder_session()
195 elif self._yaml_path:
196 self._save_folder_session([Path(self._yaml_path).parent])
197 self._restart_requested = True
198 self._window.destroy()
199 return {"success": True}
200 except Exception as error:
201 self._restart_requested = False
202 return {"success": False, "error": str(error)}
203 finally:
204 self._update_lock.release()
206 def get_layouts(self) -> dict:
207 """List workflow and named user layouts without exposing arbitrary file access."""
208 metadata = getattr(self._pipeline, "pipeline_metadata", {})
209 return {
210 "success": True,
211 "pipeline_name": metadata.get("name") or getattr(self._pipeline, "name", "Pipeline"),
212 "layouts": [{"id": key, "source": source, "name": name}
213 for key, source, name, _ in self._available_layouts()],
214 }
216 def check_updates(self) -> dict:
217 from .updates import collectra_repository, inspect_repository
219 if not self._update_lock.acquire(blocking=False):
220 return {"success": False, "error": "An update operation is already running."}
221 try:
222 targets = [
223 ("pipeline", getattr(self._pipeline, "path", None), "Pipeline"),
224 ("collectra", collectra_repository(), "Collectra"),
225 ]
226 return {"success": True, "restart_required": self._restart_required, "repositories": [
227 {"id": key, **inspect_repository(root, label)} for key, root, label in targets
228 ]}
229 finally:
230 self._update_lock.release()
232 def install_update(self, target: str) -> dict:
233 from .updates import collectra_repository, update_repository
235 if target not in {"pipeline", "collectra"}:
236 return {"success": False, "error": "Unknown update target."}
237 if not self._update_lock.acquire(blocking=False):
238 return {"success": False, "error": "An update operation is already running."}
239 try:
240 root, label = (
241 (getattr(self._pipeline, "path", None), "Pipeline") if target == "pipeline"
242 else (collectra_repository(), "Collectra")
243 )
244 repository = {"id": target, **update_repository(root, label)}
245 if repository["status"] == "updated":
246 self._restart_required = True
247 return {"success": True, "repository": repository}
248 finally:
249 self._update_lock.release()
251 def load_layout(self, layout_id: str) -> dict:
252 from .arrangement import ArrangementStore
254 for key, _, _, path in self._available_layouts():
255 if key == layout_id:
256 arrangement = ArrangementStore._read_file(path)
257 if arrangement is not None:
258 return {"success": True, "arrangement": arrangement}
259 return {"success": False, "error": "This layout is missing or invalid."}
260 return {"success": False, "error": "Layout not found."}
262 def reveal_layout(self, layout_id: str) -> dict:
263 """Reveal a workflow or custom layout in the system file manager."""
264 for key, _, _, path in self._available_layouts():
265 if key == layout_id:
266 return self.reveal_in_folder(str(path.resolve()))
267 return {"success": False, "error": "Layout not found."}
269 def save_named_layout(self, name: str, arrangement: dict) -> dict:
270 from .arrangement import sanitize_arrangement
272 clean = sanitize_arrangement(arrangement)
273 if clean is None or self._arrangement is None:
274 return {"success": False, "error": "No valid layout to save."}
275 try:
276 self._arrangement.save_named(name, clean)
277 except (ValueError, OSError) as error:
278 return {"success": False, "error": str(error)}
279 return {"success": True}
281 def activate_layout(self, layout_id: str) -> dict:
282 """Persist a chosen layout without touching the loaded results."""
283 result = self.load_layout(layout_id)
284 if not result["success"] or self._arrangement is None:
285 return result if not result["success"] else {"success": False, "error": "No pipeline open."}
286 try:
287 self._arrangement._write_file(
288 self._arrangement.path, result["arrangement"], raise_errors=True
289 )
290 except OSError as error:
291 return {"success": False, "error": str(error)}
292 return {"success": True}
294 def get_initial_items(self) -> dict:
295 """Return startup items once the JavaScript bridge is ready."""
296 return self._initial_items or {"success": True, "provided": False}
298 def _load_input_paths(self, inputs) -> dict:
299 """Resolve existing results without creating or rewriting any files."""
300 seen = set()
301 for value in inputs:
302 path = Path(value).expanduser().resolve()
303 if not path.exists():
304 raise FileNotFoundError(f"GUI input does not exist: {path}")
305 directory = path.parent if path.is_file() else path
306 if (directory / "results.yaml").is_file():
307 candidates = [directory]
308 elif path.is_dir():
309 candidates = [
310 child
311 for child in sorted(path.iterdir())
312 if child.is_dir() and child.suffix == f".{self._extension}"
313 ]
314 else:
315 raise ValueError(
316 f"GUI input must belong to a folder containing results.yaml: {path}"
317 )
318 matched = False
319 for candidate in candidates:
320 scan = self._scan_collectra_folder(candidate)
321 if scan is None:
322 continue
323 matched = True
324 candidate = candidate.resolve()
325 if candidate in seen:
326 continue
327 seen.add(candidate)
328 self._collectra_folders.append(
329 {
330 "name": candidate.name,
331 "path": str(candidate),
332 **scan,
333 }
334 )
335 if not matched:
336 raise ValueError(f"No GUI results found in: {path}")
337 self._parent_folder = str(Path(self._collectra_folders[0]["path"]).parent)
338 self._save_folder_session()
339 return {**self._folder_list_result(), "provided": True, "mode": "parent"}
341 @property
342 def _extension(self) -> str:
343 """Use the extension owned by the live pipeline."""
344 return self._pipeline.ext
346 def set_window(self, window):
347 """Store the window reference for use in dialogs."""
348 self._window = window
350 def reveal_in_folder(self, path: str) -> dict:
351 """Reveal a file in the OS file manager, selected within its folder.
353 Args:
354 path: Absolute path to reveal (e.g. a specimen's results.yaml
355 file, selected within its .grapto folder).
357 Returns:
358 dict with 'success' or 'error'
359 """
360 import subprocess
362 target = Path(path)
363 if not target.exists():
364 return {"success": False, "error": f"Path not found: {path}"}
366 system = platform.system()
367 if system == "Darwin":
368 # -R reveals + selects the item in its *parent's* Finder window —
369 # actual "Reveal in Finder" behavior — not the folder's own contents.
370 command = ["open", "-R", str(target)]
371 elif system == "Windows":
372 # Explorer's reveal+select is one combined `/select,<path>` argument.
373 command = ["explorer", f"/select,{target}"]
374 else:
375 # No universal reveal+select convention across Linux desktop
376 # environments — opening the parent folder is the closest
377 # approximation available via a single xdg-open call.
378 command = ["xdg-open", str(target.parent)]
380 try:
381 subprocess.run(command)
382 return {"success": True}
383 except Exception as e:
384 return {"success": False, "error": str(e)}
386 def _scan_collectra_folder(self, folder_path: Path) -> dict | None:
387 """
388 Scan a folder for YAML and image files.
390 Args:
391 folder_path: Path to the folder to scan
393 Returns:
394 dict with yaml_path and optional image_path, or None without results
395 """
396 results_path = folder_path / "results.yaml"
397 image_path = None
398 image_extensions = [f".{ext.lower()}" for ext in ImageFormat.__members__.keys()]
400 for filename in sorted(folder_path.iterdir(), reverse=True):
401 if not filename.is_file():
402 continue
404 if filename.suffix.lower() in image_extensions and image_path is None:
405 image_path = filename
407 if image_path:
408 break
410 if not results_path.is_file():
411 return None
413 return {
414 "yaml_path": str(results_path),
415 "image_path": str(image_path) if image_path else "",
416 }
418 def select_folder(self) -> dict:
419 """
420 Open native folder dialog and scan for YAML/image files.
422 Returns:
423 dict with folder_path, yaml_path, image_path (all absolute paths)
424 """
426 if self._window is None:
427 return {"success": False, "error": "Window not initialized"}
429 try:
430 result = self._window.create_file_dialog(
431 dialog_type=webview.FileDialog.FOLDER
432 )
434 if not result or len(result) == 0:
435 return {"success": False, "error": "No folder selected"}
437 folder_path = Path(result[0])
439 scan_result = self._scan_collectra_folder(folder_path)
440 if scan_result is None:
441 return {
442 "success": False,
443 "error": "No results.yaml found in folder",
444 }
446 return {
447 "success": True,
448 "folder_path": str(folder_path),
449 **scan_result,
450 }
452 except Exception as e:
453 return {"success": False, "error": str(e)}
455 def select_parent_folder(self) -> dict:
456 """
457 Open native folder dialog and scan for subdirectories with the specified extension.
459 Returns:
460 dict with parent_path and list of folder names/indices
461 """
462 if self._window is None:
463 return {"success": False, "error": "Window not initialized"}
465 try:
466 result = self._window.create_file_dialog(
467 dialog_type=webview.FileDialog.FOLDER
468 )
470 if not result or len(result) == 0:
471 return {"success": False, "error": "No folder selected"}
473 parent_path = Path(result[0])
474 return self._scan_parent_folder(parent_path)
476 except Exception as e:
477 return {"success": False, "error": str(e)}
479 def select_collectra_path(self) -> dict:
480 """
481 Open native folder dialog and load either a single collectra folder
482 (if it matches the extension) or a parent folder of collectra subfolders.
484 Returns:
485 dict with mode "single" or "parent" and the relevant data.
486 """
487 if self._window is None:
488 return {"success": False, "error": "Window not initialized"}
490 try:
491 result = self._window.create_file_dialog(
492 dialog_type=webview.FileDialog.FOLDER
493 )
495 if not result or len(result) == 0:
496 return {"success": False, "error": "No folder selected"}
498 selected_path = Path(result[0])
500 # A previously-viewed page's yaml path is otherwise never cleared,
501 # and _collectra_dirs() searches near it — so without this, asset
502 # lookups (get_logo, get_theme_css) can silently keep matching a
503 # file from whatever project was open before this new selection.
504 self._yaml_path = None
505 self._collectra_file = None
506 self._graph = None
508 if selected_path.suffix == f".{self._extension}":
509 scan_result = self._scan_collectra_folder(selected_path)
510 if scan_result is None:
511 return {
512 "success": False,
513 "error": "No YAML or image file found in folder",
514 }
515 self._collectra_folders = []
516 self._parent_folder = str(selected_path.parent)
517 self._save_folder_session([selected_path])
518 return {
519 "success": True,
520 "mode": "single",
521 "folder_path": str(selected_path),
522 "yaml_path": scan_result["yaml_path"],
523 "image_path": scan_result["image_path"],
524 }
526 parent_result = self._scan_parent_folder(selected_path)
527 if parent_result.get("success"):
528 parent_result["mode"] = "parent"
529 return parent_result
531 except Exception as e:
532 return {"success": False, "error": str(e)}
534 def _scan_parent_folder(self, parent_path: Path) -> dict:
535 self._parent_folder = str(parent_path)
536 self._collectra_folders = []
538 for entry in sorted(parent_path.iterdir()):
539 if not entry.is_dir():
540 continue
541 if not entry.suffix == f".{self._extension}":
542 continue
544 scan_result = self._scan_collectra_folder(entry)
545 if scan_result is not None:
546 self._collectra_folders.append(
547 {
548 "name": entry.name,
549 "path": str(entry),
550 "yaml_path": scan_result["yaml_path"],
551 "image_path": scan_result["image_path"],
552 }
553 )
555 self._save_folder_session()
556 return self._folder_list_result()
558 def _folder_list_result(self) -> dict:
559 folders = [
560 {
561 "name": f["name"],
562 "path": f["path"],
563 "yaml_path": f["yaml_path"],
564 "index": i,
565 }
566 for i, f in enumerate(self._collectra_folders)
567 ]
569 # Compute aggregate label statistics across all folders
570 self._global_label_counts = {}
571 for folder in self._collectra_folders:
572 try:
573 collectra_file = CollectraFile.from_data(Path(folder["path"]))
574 temp_graph = CollectraGraph.from_collectra_file(collectra_file)
575 label_counts = temp_graph.count_nodes_by_label(
576 type_filter="collectra.ImageCrop"
577 )
578 # Accumulate unique labels
579 self._global_labels.update(temp_graph.get_unique_labels())
580 # Accumulate counts
581 for label, count in label_counts.items():
582 self._global_label_counts[label] = (
583 self._global_label_counts.get(label, 0) + count
584 )
585 except Exception:
586 # Skip folders that can't be loaded
587 pass
589 return {
590 "success": True,
591 "parent_path": self._parent_folder,
592 "folders": folders,
593 "global_label_counts": self._global_label_counts,
594 "global_total": sum(self._global_label_counts.values()),
595 "ext": self._extension,
596 }
598 def load_collectra_folder(self, index: int) -> dict:
599 """
600 Load a folder by its index from the previously scanned parent folder.
602 Args:
603 index: Index of the folder in the _collectra_folders list
605 Returns:
606 dict with folder_name, yaml_path, image_path
607 """
608 if not self._collectra_folders:
609 return {
610 "success": False,
611 "error": "No folders loaded. Call select_parent_folder first.",
612 }
614 if index < 0 or index >= len(self._collectra_folders):
615 return {"success": False, "error": f"Invalid folder index: {index}"}
617 folder = self._collectra_folders[index]
618 return {
619 "success": True,
620 "folder_name": folder["name"],
621 "folder_path": folder["path"],
622 "yaml_path": folder["yaml_path"],
623 "image_path": folder["image_path"],
624 }
626 def get_active_node_ids(self, page_index: int) -> dict:
627 """Return the labels, real types, and real result ids that belong to a page.
629 `results.yaml` indexes each node by an internal id (a label plus a
630 generated suffix for anything not an original source node), but the
631 The pipeline DAG uses the plain label as its node id.
632 Correlating against the DAG must go through `label`, not `id`.
634 The frontend greys out and disables every DAG node whose id is NOT in
635 active_ids, can use node_types[label] to show the node's real type,
636 and node_ids[label] to show the real results.yaml id(s) — a list,
637 not a single value, since a detector-type node (e.g.
638 collectra.ObjectDetectionYOLO) declares one output label in the
639 pipeline but can produce any number of real instances per page
640 (one per detected object), all sharing that same label.
641 node_parents supplies each instance's actual direct parent ids so
642 lineage lists do not include unrelated instances of the same label.
643 """
644 if not self._collectra_folders:
645 return {"success": False, "error": "No folders loaded."}
646 if page_index < 0 or page_index >= len(self._collectra_folders):
647 return {"success": False, "error": f"Invalid page index: {page_index}"}
649 folder = self._collectra_folders[page_index]
650 try:
651 if (
652 self._graph is not None
653 and self._yaml_path is not None
654 and Path(self._yaml_path).resolve()
655 == Path(folder["yaml_path"]).resolve()
656 ):
657 graph = self._graph
658 else:
659 collectra_file = CollectraFile.from_data(Path(folder["path"]))
660 graph = CollectraGraph.from_collectra_file(collectra_file)
661 node_types: dict[str, str] = {}
662 node_ids: dict[str, list[str]] = {}
663 crop_labels = set()
664 for node_id in graph.nodes:
665 node = graph.get_node(node_id)
666 if node and node.label:
667 node_types[node.label] = node.type
668 node_ids.setdefault(node.label, []).append(node_id)
669 if isinstance(node, CollectraAnnotationNode):
670 crop_labels.add(node.label)
671 return {
672 "success": True,
673 "active_ids": list(node_types.keys()),
674 "node_types": node_types,
675 "node_ids": node_ids,
676 "node_parents": {
677 node_id: graph.parents(node_id)
678 for node_id in graph.nodes if graph.get_node(node_id)
679 },
680 "crop_labels": sorted(crop_labels),
681 }
682 except Exception as e:
683 return {"success": False, "error": str(e)}
685 def get_image_base64(self, image_path: str) -> dict:
686 """
687 Read an image file and return its base64 data.
689 Args:
690 image_path: Absolute path to the image file
692 Returns:
693 dict with base64-encoded data URI
694 """
695 try:
696 extension = Path(image_path).suffix.lower()
697 mime_type = ImageFormat[extension[1:].upper()].value
699 with open(image_path, "rb") as f:
700 image_data = f.read()
702 base64_data = base64.b64encode(image_data).decode("utf-8")
704 return {"success": True, "data": f"data:{mime_type};base64,{base64_data}"}
706 except Exception as e:
707 return {"success": False, "error": str(e)}
709 def _display_context(self, rgb_view: int = 0):
710 from collectra.display import DisplayContext
712 if self._graph is None or self._yaml_path is None:
713 raise ValueError("No page loaded")
714 records = {
715 node_id: {
716 **self._graph.get_node(node_id).model_dump(),
717 "label": self._graph.get_node(node_id).label,
718 }
719 for node_id in self._graph.nodes
720 if self._graph.get_node(node_id)
721 }
722 return DisplayContext(records, Path(self._yaml_path).parent, rgb_view=rgb_view)
724 def get_artefact_display(self, node_id: str = "", rgb_view: int = 0) -> dict:
725 """Ask the actual artefact class how it should be presented."""
726 try:
727 context = self._display_context(rgb_view=rgb_view)
728 if not node_id:
729 from collectra.types.images import Image, ImageCrop
730 from collectra.utils import load_class_from_string
732 for candidate in context.records:
733 try:
734 cls = load_class_from_string(context.records[candidate]["type"])
735 except (ImportError, AttributeError):
736 continue
737 if issubclass(cls, Image) and not issubclass(cls, ImageCrop):
738 node_id = candidate
739 break
740 node_id = self._graph.resolve_id(node_id)
741 item = context.artefact(node_id)
742 view = item.display(context)
743 return {
744 "success": True,
745 "view": view,
746 "id": node_id,
747 "type": context.records[node_id]["type"],
748 }
749 except Exception as e:
750 return {"success": False, "error": str(e)}
752 def rotate_image_clockwise(self, node_id: str) -> dict:
753 """Persist a clockwise quarter-turn for an image and descendant images."""
754 from collectra.types.images import Image, Orientation
755 from collectra.types.links import Link
756 from collectra.utils import load_class_from_string
758 changes = []
759 try:
760 context = self._display_context()
761 item = context.artefact(self._graph.resolve_id(node_id))
762 if isinstance(item, Link):
763 item = item.resolve()
764 if not isinstance(item, Image):
765 raise ValueError("Only image artefacts can be rotated")
766 pending, seen = [item.id], set()
767 while pending:
768 current = pending.pop()
769 if current in seen:
770 continue
771 seen.add(current)
772 pending.extend(self._graph.children(current))
773 node = self._graph.get_node(current)
774 if node and issubclass(load_class_from_string(node.type), Image):
775 orientation = Orientation.from_string(node.orientation)
776 next_orientation = Orientation(
777 (orientation.value + 1) % 4
778 ).to_string()
779 changes.append((node, node.orientation, next_orientation))
780 for node, previous, updated in changes:
781 node.orientation = updated
782 self._save_collectra_file()
783 return {"success": True, "updated_ids": [node.id for node, _, _ in changes]}
784 except Exception as error:
785 for node, previous, updated in changes:
786 node.orientation = previous
787 return {"success": False, "error": str(error)}
789 def update_segmentation_mask(
790 self, node_id: str, strokes: list, results_path: str, orientation: str
791 ) -> dict:
792 """Save strokes to their original results page, even after navigation."""
793 from collectra.display import DisplayContext
794 from collectra.displays.image_segmentation import segmentation
795 from collectra.gui.segmentation import paint_mask
796 from collectra.types.links import Link
798 try:
799 with self._text_save_lock:
800 target = Path(results_path).resolve()
801 if target.name != "results.yaml" or not target.is_file():
802 raise ValueError("Mask edits must target a results.yaml file")
803 collectra_file = CollectraFile.from_data(target.parent)
804 graph = CollectraGraph.from_collectra_file(collectra_file)
805 records = {
806 key: {**graph.get_node(key).model_dump(), "label": graph.get_node(key).label}
807 for key in graph.nodes if graph.get_node(key)
808 }
809 context = DisplayContext(records, target.parent)
810 item = context.artefact(graph.resolve_id(node_id))
811 if isinstance(item, Link):
812 item = item.resolve()
813 changed = paint_mask(item, strokes, orientation)
814 return {"success": True, "changed": changed, "segmentation": segmentation(context, item)}
815 except Exception as error:
816 return {"success": False, "error": str(error)}
818 def get_display_annotations(self, node_id: str) -> dict:
819 try:
820 context = self._display_context()
821 image = context.artefact(self._graph.resolve_id(node_id))
822 from collectra.types.links import Link
824 if type(image) is Link:
825 image = image.resolve()
826 return {"success": True, "rows": context.annotations(image)}
827 except Exception as e:
828 return {"success": False, "error": str(e)}
830 def update_display_text(
831 self, node_id: str, content: str, results_path: str | None = None
832 ) -> dict:
833 """Save changed text to its captured page, even after navigation."""
834 try:
835 from collectra.display import DisplayContext
837 with self._text_save_lock:
838 target = Path(results_path or self._yaml_path).resolve()
839 if target.name != "results.yaml":
840 raise ValueError("Text edits must target a results.yaml file")
841 if self._yaml_path and target == Path(self._yaml_path).resolve():
842 graph, collectra_file = self._graph, self._collectra_file
843 else:
844 collectra_file = CollectraFile.from_data(target.parent)
845 graph = CollectraGraph.from_collectra_file(collectra_file)
846 records = {
847 key: {**graph.get_node(key).model_dump(), "label": graph.get_node(key).label}
848 for key in graph.nodes if graph.get_node(key)
849 }
850 context = DisplayContext(records, target.parent)
851 item = context.artefact(graph.resolve_id(node_id))
852 view = item.display(context)
853 if view.get("kind") != "text" or not view.get("editable"):
854 raise ValueError("This artefact does not expose editable text")
855 target_id = view["target_id"]
856 path = context.text_path(target_id)
857 if path:
858 if path.read_text(encoding="utf-8") == content:
859 return {"success": True, "changed": False}
860 path.write_text(content, encoding="utf-8")
861 else:
862 if graph.get_data(target_id) == content:
863 return {"success": True, "changed": False}
864 previous = graph.get_data(target_id)
865 graph.set_data(target_id, content)
866 try:
867 collectra_file.data = graph.to_data()
868 collectra_file.save()
869 except Exception:
870 graph.set_data(target_id, previous)
871 collectra_file.data = graph.to_data()
872 raise
873 return {"success": True, "changed": True}
874 except Exception as e:
875 return {"success": False, "error": str(e)}
877 def load_yaml(self, path: str) -> dict:
878 with self._text_save_lock:
879 return self._load_yaml(path)
881 def _load_yaml(self, path: str) -> dict:
882 """
883 Load a YAML file and build the annotation graph.
885 Args:
886 path: Path to the YAML file
888 Returns:
889 dict with 'success', 'node_count', 'edge_count', or 'error'
890 """
891 try:
892 collectra_path = Path(path)
893 directory = (
894 collectra_path if collectra_path.is_dir() else collectra_path.parent
895 )
896 collectra_file = CollectraFile.from_data(directory)
897 graph = CollectraGraph.from_collectra_file(collectra_file)
898 self._collectra_file = collectra_file
899 self._graph = graph
900 previous_path = self._yaml_path
901 self._yaml_path = str(self._collectra_file.results_path)
902 if self._session_cache is not None and previous_path != self._yaml_path:
903 self._session_cache.write_active(
904 str(self._collectra_file.results_path.parent.resolve())
905 )
906 if not self._collectra_folders:
907 self._save_folder_session([self._collectra_file.results_path.parent])
909 # Accumulate labels from this graph
910 self._global_labels.update(self._graph.get_unique_labels())
912 return {"success": True, "path": self._yaml_path}
913 except Exception as e:
914 return {"success": False, "error": str(e)}
916 def get_display_value(self, node_id: str) -> dict:
917 """
918 Compute display value for an annotation.
920 Args:
921 node_id: The annotation ID
923 Returns:
924 dict with 'value', 'source_id', 'reason', or 'error'
925 """
926 if self._graph is None:
927 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
929 try:
930 result = self._graph.compute_display_value(self._graph.resolve_id(node_id))
931 return {
932 "success": True,
933 "value": result.value,
934 "source_id": result.source_id,
935 "crop_region": result.crop_region,
936 "reason": result.reason,
937 }
938 except Exception as e:
939 return {"success": False, "error": str(e)}
941 def get_page_texts(self) -> dict:
942 """Return markdown and TEI content for the current page folder."""
943 if self._yaml_path is None:
944 return {"success": False, "error": "No page loaded."}
945 page_dir = Path(self._yaml_path).parent
946 result: dict = {"success": True}
947 for key, filename in (("markdown", "markdown.md"), ("tei", "tei.xml")):
948 p = page_dir / filename
949 if p.exists():
950 result[key] = p.read_text(encoding="utf-8")
951 result[f"{key}_found"] = True
952 else:
953 result[key] = None
954 result[f"{key}_found"] = False
955 return result
957 def update_page_text(self, kind: str, content: str) -> dict:
958 """Write markdown or TEI content back to the current page's folder.
960 Args:
961 kind: "markdown" or "tei"
962 content: the full text to write
964 Returns:
965 dict with 'success' or 'error'
966 """
967 if self._yaml_path is None:
968 return {"success": False, "error": "No page loaded."}
970 filenames = {"markdown": "markdown.md", "tei": "tei.xml"}
971 filename = filenames.get(kind)
972 if filename is None:
973 return {"success": False, "error": f"Unknown kind: {kind}"}
975 try:
976 page_dir = Path(self._yaml_path).parent
977 (page_dir / filename).write_text(content, encoding="utf-8")
978 return {"success": True}
979 except Exception as e:
980 return {"success": False, "error": str(e)}
982 def get_all_nodes(self) -> dict:
983 """
984 Get all node IDs in the graph.
986 Returns:
987 dict with 'nodes' list or 'error'
988 """
989 if self._graph is None:
990 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
992 return {"success": True, "nodes": self._graph.nodes}
994 def get_node_info(self, node_id: str) -> dict:
995 """
996 Get detailed information about a node.
998 Args:
999 node_id: The annotation ID
1001 Returns:
1002 dict with node type, data, children, parents, or 'error'
1003 """
1004 if self._graph is None:
1005 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1007 node_id = self._graph.resolve_id(node_id)
1008 if node_id not in self._graph.nodes:
1009 return {"success": False, "error": f"Node '{node_id}' not found"}
1011 return {
1012 "success": True,
1013 "id": node_id,
1014 "type": self._graph.get_type(node_id),
1015 "data": self._graph.get_data(node_id),
1016 "children": self._graph.children(node_id),
1017 "parents": self._graph.parents(node_id),
1018 }
1020 def get_nodes_by_type(self, type_filter: str) -> dict:
1021 """
1022 Get all nodes containing a type substring.
1024 Args:
1025 type_filter: Substring to match in node types (e.g., "ImageCrop", "Text")
1027 Returns:
1028 dict with 'nodes' list of matching node IDs
1029 """
1030 if self._graph is None:
1031 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1033 matching = [
1034 node_id
1035 for node_id in self._graph.nodes
1036 if type_filter in self._graph.get_type(node_id)
1037 ]
1039 return {"success": True, "nodes": matching, "count": len(matching)}
1041 def get_available_labels(self) -> dict:
1042 """
1043 Get all unique labels accumulated across all loaded files.
1045 Returns:
1046 dict with 'success' and 'labels' list (sorted)
1047 """
1048 try:
1049 labels = sorted(self._global_labels)
1050 return {"success": True, "labels": labels}
1051 except Exception as e:
1052 return {"success": False, "error": str(e)}
1054 def get_label_statistics(self) -> dict:
1055 """
1056 Return counts of nodes per label in current graph.
1058 Returns:
1059 dict with 'success', 'label_counts', and 'total'
1060 """
1061 if self._graph is None:
1062 return {"success": False, "error": "No graph loaded"}
1064 # Count only ImageCrop types (annotations visible in grid)
1065 label_counts = self._graph.count_nodes_by_label(
1066 type_filter="collectra.ImageCrop"
1067 )
1069 return {
1070 "success": True,
1071 "label_counts": label_counts,
1072 "total": sum(label_counts.values()),
1073 }
1075 def get_global_label_statistics(self) -> dict:
1076 """
1077 Return aggregated label counts across all files in parent folder.
1079 Returns:
1080 dict with 'success', 'label_counts', and 'total'
1081 """
1082 return {
1083 "success": True,
1084 "label_counts": self._global_label_counts,
1085 "total": sum(self._global_label_counts.values()),
1086 }
1088 def get_all_nodes_for_grid(self) -> dict:
1089 """
1090 Return all nodes formatted for AG Grid rows.
1092 Returns:
1093 dict with 'rows' list containing node data for grid display
1094 """
1095 if self._graph is None:
1096 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1098 rows = []
1099 for node_id in self._graph.nodes:
1100 node_display_value: NodeDisplayValue = self._graph.compute_display_value(
1101 node_id
1102 )
1103 rows.append(
1104 {
1105 "id": node_id,
1106 "type": self._graph.get_type(node_id),
1107 "data": str(self._graph.get_data(node_id)),
1108 "name": self._graph.get_name(node_id),
1109 "label": self._graph.get_label(node_id),
1110 "displayValue": node_display_value.value,
1111 "crop_region": node_display_value.crop_region,
1112 "displaySourceId": node_display_value.source_id,
1113 "reason": node_display_value.reason,
1114 "parents": ", ".join(self._graph.parents(node_id)),
1115 "children": ", ".join(self._graph.children(node_id)),
1116 "locked": node_display_value.locked,
1117 }
1118 )
1120 return {"success": True, "rows": rows}
1122 def update_node_data(
1123 self, node_id: str, new_data: str, crop_id: str | None = None
1124 ) -> dict:
1125 """
1126 Update the data of a specific node.
1128 Args:
1129 node_id: The annotation ID
1130 new_data: New data to set for the node
1132 Returns:
1133 dict with 'success' or 'error'
1134 """
1135 try:
1136 if self._graph is None:
1137 raise ValueError("No graph loaded. Call load_yaml first.")
1138 # node_id may be "" when creating a new Text node via crop_id — only
1139 # resolve a real, non-empty id (an empty label would never match).
1140 resolved_id = self._graph.resolve_id(node_id) if node_id else node_id
1141 self._graph.set_data(resolved_id, new_data, crop_id)
1142 self._save_collectra_file()
1143 return self.get_all_nodes_for_grid()
1144 except Exception as e:
1145 return {"success": False, "error": str(e)}
1147 def update_node_coordinates(
1148 self, node_id: str, crop_region: dict, view_id: str = ""
1149 ) -> dict:
1150 """
1151 Update the crop region coordinates of a specific node.
1153 Args:
1154 node_id: The annotation ID
1155 crop_region: dict with x_center, y_center, width_relative, height_relative
1157 Returns:
1158 dict with 'success' and updated rows, or 'error'
1159 """
1160 try:
1161 if self._graph is None:
1162 raise ValueError("No graph loaded. Call load_yaml first.")
1163 if view_id:
1164 context = self._display_context()
1165 crop_region = context.region(
1166 context.artefact(view_id), crop_region, inverse=True
1167 )
1168 self._graph.set_crop_region(self._graph.resolve_id(node_id), crop_region)
1169 self._save_collectra_file()
1170 return self.get_all_nodes_for_grid()
1171 except Exception as e:
1172 return {"success": False, "error": str(e)}
1174 def _next_annotation_id(self, label: str) -> str:
1175 """Matches the pipeline's scheme (label+N) — reuses the smallest
1176 suffix a deletion freed up, rather than always growing past the
1177 highest one ever used. Shared with preview_annotation_id."""
1178 existing_numbers = set()
1179 for node_id in self._graph.nodes:
1180 node = self._graph.get_node(node_id)
1181 if node and node.label == label:
1182 suffix = node_id[len(label) :]
1183 if suffix.isdigit():
1184 existing_numbers.add(int(suffix))
1185 n = 1
1186 while n in existing_numbers:
1187 n += 1
1188 return f"{label}{n}"
1190 def preview_annotation_id(self, label: str) -> dict:
1191 """What create_annotation would name the next box with this label, without creating anything."""
1192 if self._graph is None:
1193 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1194 return {"success": True, "id": self._next_annotation_id(label)}
1196 def create_annotation(
1197 self,
1198 crop_region: dict,
1199 label: str,
1200 parent_id: str,
1201 image_path: str = "",
1202 name: str = "",
1203 crop_id: str = "",
1204 view_id: str = "",
1205 ) -> dict:
1206 """
1207 Create a new ImageCrop annotation with a Text child.
1209 Args:
1210 crop_region: dict with x_center, y_center, width_relative, height_relative
1211 name: optional per-instance display name, distinct from label
1212 crop_id: explicit node id; auto-numbered (label + N) when blank
1214 Returns:
1215 dict with updated grid data or 'error'
1216 """
1217 if self._graph is None:
1218 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1220 if not image_path:
1221 return {
1222 "success": False,
1223 "error": "Image path is required to create annotation.",
1224 }
1226 try:
1227 if view_id:
1228 context = self._display_context()
1229 image = context.artefact(view_id)
1230 crop_region = context.region(image, crop_region, inverse=True)
1231 image_path = str(image.get_path())
1232 parent_id = parent_id or image.id
1233 # Find root Image node
1234 root_image_id = parent_id if view_id else None
1235 for node_id in self._graph.nodes:
1236 node_type = self._graph.get_type(node_id)
1237 if "Image" in node_type and "ImageCrop" not in node_type:
1238 root_image_id = node_id
1239 break
1241 if root_image_id is None:
1242 return {"success": False, "error": "No root Image node found in graph."}
1244 orientation = "north"
1246 if parent_id:
1247 parent_node = self._graph.get_node(parent_id)
1248 if parent_node:
1249 orientation = parent_node.orientation
1251 crop_id = crop_id.strip() or self._next_annotation_id(label)
1252 if crop_id in self._graph.nodes:
1253 return {"success": False, "error": f"Id '{crop_id}' already exists."}
1255 if self._yaml_path is None:
1256 raise ValueError("YAML path is not set.")
1258 asset_path = Path(image_path)
1259 if view_id:
1260 try:
1261 asset_path = asset_path.relative_to(context.directory)
1262 except ValueError:
1263 pass
1264 else:
1265 asset_path = Path(asset_path.name)
1266 # Create ImageCrop node
1267 crop_data = {
1268 "label": label,
1269 "type": "collectra.ImageCrop",
1270 "id": crop_id,
1271 "parents": parent_id if parent_id else root_image_id,
1272 "orientation": orientation,
1273 "data": str(asset_path),
1274 "name": name,
1275 **crop_region,
1276 }
1277 self._graph.add_node(crop_data)
1278 self._global_labels.add(label)
1279 # Update global label counts if in parent folder mode
1280 if self._collectra_folders:
1281 self._global_label_counts[label] = (
1282 self._global_label_counts.get(label, 0) + 1
1283 )
1284 # Save and return updated grid
1285 self._save_collectra_file()
1286 result = self.get_all_nodes_for_grid()
1287 # Include updated global stats in response
1288 if self._collectra_folders:
1289 result["global_label_counts"] = self._global_label_counts
1290 result["global_total"] = sum(self._global_label_counts.values())
1291 return result
1293 except Exception as e:
1294 return {"success": False, "error": str(e)}
1296 def _node_and_descendants(self, node_id: str) -> list[str]:
1297 """The node followed by every descendant (any type, any depth),
1298 breadth first, each id once."""
1299 ids = []
1300 queue = [node_id]
1301 seen = {node_id}
1302 while queue:
1303 current = queue.pop(0)
1304 ids.append(current)
1305 for child_id in self._graph.children(current):
1306 if child_id not in seen:
1307 seen.add(child_id)
1308 queue.append(child_id)
1309 return ids
1311 def search_artefacts(self, query: str) -> dict:
1312 """
1313 Find the artefacts on the open page that contain ``query``
1314 (case-insensitive) in their content: the text of a Text artefact or
1315 a custom name. DAG labels and ids are not searched.
1317 Returns:
1318 dict with 'matches': one entry per artefact, in page order, each
1319 with id, DAG label (to place it in the tabs), type, the first
1320 field that matched (text, name) and a snippet around the match
1321 """
1322 if self._graph is None:
1323 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1325 needle = (query or "").strip().lower()
1326 if not needle:
1327 return {"success": True, "matches": []}
1329 from collectra.types.texts import Text
1330 from collectra.utils import load_class_from_string
1332 context = self._display_context()
1333 matches = []
1334 for node_id in self._nodes_top_down():
1335 node = self._graph.get_node(node_id)
1336 if node is None:
1337 continue
1338 fields = []
1339 try:
1340 if issubclass(load_class_from_string(node.type), Text):
1341 path = context.text_path(node_id)
1342 if path:
1343 text = path.read_text(encoding="utf-8")
1344 elif context.records.get(node_id, {}).get("path"):
1345 text = (
1346 "" # its file is missing: nothing to search, not its path
1347 )
1348 else:
1349 text = node.data
1350 fields.append(("text", str(text or "")))
1351 except Exception:
1352 # One artefact that cannot be read (unknown type, a type that is not
1353 # a class, a file that is not text) must not stop the whole search.
1354 pass
1355 fields.append(("name", node.name or ""))
1356 for field, value in fields:
1357 found = re.search(re.escape(needle), value, re.IGNORECASE)
1358 position = found.start() if found else -1
1359 if position != -1:
1360 matches.append(
1361 {
1362 "id": node_id,
1363 "label": node.label,
1364 "type": node.type,
1365 "field": field,
1366 "snippet": self._snippet(
1367 value, position, len(found.group())
1368 ),
1369 }
1370 )
1371 break
1372 # Walk the matches the way the pipeline reads; labels it does not know
1373 # (made by hand) come last. The sort is stable, so within one label the
1374 # order of the page is kept.
1375 ranks = self._pipeline_label_ranks()
1376 matches.sort(key=lambda match: ranks.get(match["label"], len(ranks)))
1377 return {"success": True, "matches": matches}
1379 def _pipeline_label_ranks(self) -> dict[str, int]:
1380 from .pipeline_order import pipeline_label_ranks
1382 try:
1383 return pipeline_label_ranks(self._pipeline.gui_graph())
1384 except Exception:
1385 return {}
1387 def _nodes_top_down(self) -> list[str]:
1388 """Every node, each followed by its descendants (the image, then its
1389 crops, then their texts), so results read in the order of the page."""
1390 ordered, seen = [], set()
1392 def visit(node_id: str) -> None:
1393 if node_id in seen:
1394 return
1395 seen.add(node_id)
1396 ordered.append(node_id)
1397 for child_id in self._graph.children(node_id):
1398 visit(child_id)
1400 for node_id in self._graph.nodes:
1401 if not self._graph.parents(node_id):
1402 visit(node_id)
1403 for node_id in self._graph.nodes: # anything only reachable around a cycle
1404 visit(node_id)
1405 return ordered
1407 @staticmethod
1408 def _snippet(value: str, position: int, length: int, context: int = 40) -> str:
1409 """The match with a little text either side, trimmed with ellipses."""
1410 start = max(0, position - context)
1411 end = min(len(value), position + length + context)
1412 snippet = " ".join(value[start:end].split())
1413 return ("…" if start > 0 else "") + snippet + ("…" if end < len(value) else "")
1415 def get_delete_preview(self, node_id: str) -> dict:
1416 """
1417 List what delete_annotation would remove along with a node, without
1418 changing anything.
1420 Args:
1421 node_id: The annotation ID (or DAG label) about to be deleted
1423 Returns:
1424 dict with the resolved 'id' and 'children' (each with id, label
1425 and type, nearest first), or 'error' on failure
1426 """
1427 if self._graph is None:
1428 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1430 node_id = self._graph.resolve_id(node_id)
1431 if node_id not in self._graph.nodes:
1432 return {"success": False, "error": f"Node '{node_id}' not found"}
1434 children = []
1435 seen = {node_id}
1437 def walk(current: str, depth: int) -> None:
1438 for child_id in self._graph.children(current):
1439 if child_id in seen:
1440 continue
1441 seen.add(child_id)
1442 node = self._graph.get_node(child_id)
1443 children.append(
1444 {
1445 "id": child_id,
1446 "label": node.label if node else "",
1447 "type": node.type if node else "",
1448 "depth": depth,
1449 }
1450 )
1451 walk(child_id, depth + 1)
1453 walk(node_id, 1)
1454 return {
1455 "success": True,
1456 "id": node_id,
1457 "children": children,
1458 "keeps_self": not self._graph.parents(node_id),
1459 }
1461 def _remove_nodes(self, to_delete: list[str]) -> dict:
1462 """Remove the given nodes, update label counts, save and return the
1463 refreshed grid plus the removed ids."""
1464 deleted_labels = [
1465 node.label
1466 for node_id in to_delete
1467 if (node := self._graph.get_node(node_id))
1468 ]
1470 # Deepest-reached first, so a node's own children are already gone
1471 # before it is.
1472 for node_id in reversed(to_delete):
1473 self._graph.remove_node(node_id)
1475 # Update global label counts if in parent folder mode
1476 if self._collectra_folders:
1477 for label in deleted_labels:
1478 if label in self._global_label_counts:
1479 self._global_label_counts[label] -= 1
1480 if self._global_label_counts[label] <= 0:
1481 del self._global_label_counts[label]
1483 self._save_collectra_file()
1484 result = self.get_all_nodes_for_grid()
1485 # Every id actually removed — the frontend needs this to close any
1486 # tab left open on a cascaded descendant.
1487 result["deleted_ids"] = to_delete
1488 if self._collectra_folders:
1489 result["global_label_counts"] = self._global_label_counts
1490 result["global_total"] = sum(self._global_label_counts.values())
1491 return result
1493 def delete_nested(self, node_id: str) -> dict:
1494 """
1495 Delete everything nested under a node but keep the node itself (used
1496 for the page image, which cannot be removed).
1498 Returns:
1499 dict with updated grid data and 'deleted_ids', or 'error'
1500 """
1501 if self._graph is None:
1502 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1504 try:
1505 node_id = self._graph.resolve_id(node_id)
1506 if node_id not in self._graph.nodes:
1507 return {"success": False, "error": f"Node '{node_id}' not found"}
1508 return self._remove_nodes(self._node_and_descendants(node_id)[1:])
1509 except Exception as e:
1510 return {"success": False, "error": str(e)}
1512 def delete_annotation(self, node_id: str) -> dict:
1513 """
1514 Delete an ImageCrop annotation and everything nested under it —
1515 cascades to its Text children, any child ImageCrops (sub-crops), and
1516 so on at any depth. Deleting a crop that still has children used to
1517 leave them behind pointing at a now-nonexistent parent id (a dangling
1518 reference the graph library then treats as a bare placeholder node,
1519 breaking id-uniqueness checks for that id ever after) — cascading is
1520 the only way to guarantee no orphans.
1522 Args:
1523 node_id: The ImageCrop annotation ID to delete
1525 Returns:
1526 dict with updated grid data on success, or 'error' on failure
1527 """
1528 if self._graph is None:
1529 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1531 try:
1532 node_id = self._graph.resolve_id(node_id)
1534 # Collect every descendant before deleting anything — walking the
1535 # graph while removing nodes from it would skip some.
1536 return self._remove_nodes(self._node_and_descendants(node_id))
1538 except Exception as e:
1539 return {"success": False, "error": str(e)}
1541 def rename_annotation(
1542 self,
1543 node_id: str,
1544 name: str,
1545 label: str | None = None,
1546 new_id: str | None = None,
1547 ) -> dict:
1548 """
1549 Sets an ImageCrop annotation's display name. label and new_id are
1550 optional; omitted, that field is left as-is.
1552 Args:
1553 node_id: The ImageCrop annotation ID to rename
1554 name: The new name
1555 label: optional new label
1556 new_id: optional new id — re-points every parent reference to match
1558 Returns:
1559 dict with updated grid data on success, or 'error' on failure
1560 """
1561 if self._graph is None:
1562 return {"success": False, "error": "No graph loaded. Call load_yaml first."}
1564 try:
1565 node_id = self._graph.resolve_id(node_id)
1566 if new_id and new_id != node_id:
1567 self._graph.rename_id(node_id, new_id)
1568 node_id = new_id
1569 if label is not None:
1570 self._graph.set_label(node_id, label)
1571 self._graph.set_name(node_id, name)
1572 self._save_collectra_file()
1573 return self.get_all_nodes_for_grid()
1575 except Exception as e:
1576 return {"success": False, "error": str(e)}
1578 def _save_collectra_file(self) -> None:
1579 """Persist GUI edits through the shared Collectra file writer."""
1580 if self._collectra_file is None or self._graph is None:
1581 raise ValueError("No Collectra file loaded to save to.")
1582 self._collectra_file.data = self._graph.to_data()
1583 self._collectra_file.save()
1585 def _collectra_dirs(self) -> list[Path]:
1586 """Return all .collectra directories to search, in priority order.
1588 Search order: pipeline path → root itself → root's children → root's parent's children.
1589 The parent search lets the user open a book sub-folder (e.g. Livy-Books21-25/)
1590 while the .collectra lives as a sibling inside the project root (LivyProject/).
1591 """
1592 dirs: list[Path] = []
1593 pipeline_path = self._pipeline.path
1594 if pipeline_path.is_dir():
1595 dirs.append(pipeline_path)
1596 roots: list[Path] = []
1597 if self._parent_folder:
1598 roots.append(Path(self._parent_folder))
1599 if self._yaml_path:
1600 roots.append(Path(self._yaml_path).parent.parent)
1601 seen = set()
1603 def _scan(directory: Path) -> None:
1604 """Add any .collectra dirs found directly inside `directory`."""
1605 if not directory.is_dir():
1606 return
1607 if directory.suffix == ".collectra" and directory not in seen:
1608 dirs.append(directory)
1609 seen.add(directory)
1610 return
1611 try:
1612 for entry in directory.iterdir():
1613 if (
1614 entry.is_dir()
1615 and entry.suffix == ".collectra"
1616 and entry not in seen
1617 ):
1618 dirs.append(entry)
1619 seen.add(entry)
1620 except (PermissionError, OSError):
1621 pass
1623 for root in roots:
1624 _scan(root) # look inside the selected folder
1625 _scan(root.parent) # look inside the parent (sibling .collectra case)
1627 return dirs
1629 def _pipeline_metadata(self) -> dict:
1630 """Return metadata retained by the live pipeline object."""
1631 return self._pipeline.pipeline_metadata.copy()
1633 def get_pipeline_graph(self) -> dict:
1634 """Return the graph already constructed by the live pipeline."""
1635 return {"success": True, **self._pipeline.gui_graph()}
1637 def get_theme_css(self) -> dict:
1638 """Load the CSS theme named in pipeline metadata, anywhere in the .collectra tree.
1640 Falls back to a file literally named theme.css, then any CSS file, when no
1641 named file is found.
1642 """
1643 named = self._pipeline_metadata().get("theme", "")
1644 for directory in self._collectra_dirs():
1645 # 1. Exact name from metadata
1646 if named:
1647 try:
1648 hits = list(directory.rglob(named))
1649 except (PermissionError, OSError):
1650 hits = []
1651 if hits:
1652 try:
1653 with open(hits[0], "r") as f:
1654 return {
1655 "success": True,
1656 "css": f.read(),
1657 "path": str(hits[0]),
1658 }
1659 except Exception as e:
1660 return {"success": False, "error": str(e)}
1662 # 2. Fallback: prefer a file named theme.css, else any CSS file
1663 try:
1664 css_files = [
1665 f for f in directory.iterdir() if f.suffix == ".css" and f.is_file()
1666 ]
1667 except (PermissionError, OSError):
1668 continue
1669 preferred = next((f for f in css_files if f.stem == "theme"), None)
1670 target = preferred or (css_files[0] if css_files else None)
1671 if target:
1672 try:
1673 with open(target, "r") as f:
1674 return {"success": True, "css": f.read(), "path": str(target)}
1675 except Exception as e:
1676 return {"success": False, "error": str(e)}
1678 # Fall back to bundled default asset
1679 asset_path = Path(get_resource_path("assets/theme.css"))
1680 if asset_path.is_file():
1681 try:
1682 with open(asset_path, "r") as f:
1683 return {"success": True, "css": f.read(), "path": str(asset_path)}
1684 except Exception as e:
1685 return {"success": False, "error": str(e)}
1687 return {"success": False, "error": "No CSS theme file found"}
1689 def _encode_logo(self, path: Path) -> dict:
1690 """Encode a logo file (PNG/SVG/etc.) as a base64 data URI with the right MIME type."""
1691 try:
1692 ext = path.suffix.lower()
1693 if ext == ".svg":
1694 mime = "image/svg+xml"
1695 else:
1696 mime = ImageFormat[ext[1:].upper()].value
1697 with open(path, "rb") as f:
1698 data = base64.b64encode(f.read()).decode("utf-8")
1699 return {"success": True, "data": f"data:{mime};base64,{data}"}
1700 except Exception as e:
1701 return {"success": False, "error": str(e)}
1703 def get_logo(self) -> dict:
1704 """Find the logo named in pipeline metadata (PNG or SVG), anywhere in the .collectra tree.
1706 Falls back to globbing for a logo-ish image when no named file is found,
1707 then to Collectra's own bundled logo when the project has none at all —
1708 callers rely on this always succeeding so a project with no logo doesn't
1709 keep showing whichever project's logo loaded previously.
1710 """
1711 named = self._pipeline_metadata().get("logo", "")
1712 image_exts = {f".{e.lower()}" for e in ImageFormat.__members__} | {".svg"}
1713 for directory in self._collectra_dirs():
1714 # 1. Exact name from metadata
1715 if named:
1716 try:
1717 hits = list(directory.rglob(named))
1718 except (PermissionError, OSError):
1719 hits = []
1720 if hits:
1721 return self._encode_logo(hits[0])
1722 # 2. Glob fallback: prefer a "logo"-named image, else any image/SVG
1723 try:
1724 candidates = [
1725 f for f in directory.rglob("*") if f.suffix.lower() in image_exts
1726 ]
1727 except (PermissionError, OSError):
1728 continue
1729 logos = [f for f in candidates if "logo" in f.stem.lower()]
1730 target = logos[0] if logos else (candidates[0] if candidates else None)
1731 if target:
1732 return self._encode_logo(target)
1734 # Fall back to bundled default asset
1735 asset_path = Path(get_resource_path("assets/CollectraLogoAndTitle.svg"))
1736 if asset_path.is_file():
1737 return self._encode_logo(asset_path)
1739 return {"success": False, "error": "No logo file found"}
1742def _set_macos_app_name(name: str) -> None:
1743 """Set the app identity before Cocoa creates its native application menu."""
1744 if platform.system() != "Darwin":
1745 return
1746 try:
1747 from Foundation import NSBundle, NSProcessInfo
1749 info = NSBundle.mainBundle().infoDictionary()
1750 info["CFBundleName"] = name
1751 info["CFBundleDisplayName"] = name
1752 NSProcessInfo.processInfo().setProcessName_(name)
1753 except Exception:
1754 # A packaged launcher also supplies these names in its Info.plist.
1755 import logging
1757 logging.getLogger(__name__).warning("Could not set macOS application name", exc_info=True)
1760def _disable_macos_tabbing() -> None:
1761 """Disable macOS Window Tab Bar for the pywebview window."""
1762 import platform
1764 if platform.system() != "Darwin":
1765 return
1766 try:
1767 from AppKit import NSApplication
1769 for win in NSApplication.sharedApplication().windows():
1770 win.setTabbingMode_(2) # NSWindowTabbingModeDisallowed = 2
1771 except Exception:
1772 pass
1775def get_resource_path(relative_path: str) -> str:
1776 """Get the path to a bundled resource, works for dev and PyInstaller."""
1777 import sys
1779 if getattr(sys, "frozen", False):
1780 # Running as PyInstaller bundle
1781 base_path = Path(getattr(sys, "_MEIPASS", ""))
1782 else:
1783 # Running in development
1784 base_path = Path(__file__).parent
1785 return str(base_path / relative_path)
1788def start(
1789 pipeline: "Collectra",
1790 debug: bool = False,
1791 inputs=(),
1792):
1793 """
1794 Create and start the pywebview window with the API.
1796 Args:
1797 pipeline: The live pipeline whose graph the GUI displays.
1798 debug: Enable developer tools
1799 inputs: Optional result files/folders to populate the sidebar at startup.
1800 """
1801 _set_macos_app_name(pipeline.name)
1802 api = GUIBackend(pipeline, inputs=inputs)
1803 html_path = get_resource_path("index.html")
1804 window = webview.create_window(
1805 title=pipeline.name,
1806 url=html_path,
1807 js_api=api,
1808 width=1200,
1809 height=800,
1810 min_size=(800, 600),
1811 )
1813 def on_started():
1814 api.set_window(window)
1815 _disable_macos_tabbing()
1816 arrange_native_menu()
1818 from .native_menu import arrange_native_menu, build_native_menu
1820 webview.start(func=on_started, debug=debug, menu=build_native_menu(window))
1821 if api._restart_requested:
1822 import os
1823 import sys
1825 arguments = [sys.executable, "-m", "collectra.main", "--pipeline",
1826 str(Path(pipeline.path).resolve() / "pipeline.yaml"), "gui"]
1827 if debug:
1828 arguments.append("--debug")
1829 # Reuse the environment and process. Startup restores saved results,
1830 # the selected page, and layout rather than the original launch inputs.
1831 os.execv(sys.executable, arguments)
1832 return window