"""``N4`` — the Project Browser: every project on disk, in one table.
Navigating by folder is how spaCR has always been used, and it is why nobody
can answer "which plates have been measured?", "which one is the 400 GB?" or
"which results no longer match the masks under them?" without opening six
windows. This screen is that answer as a list.
It computes nothing. Every column comes from :mod:`spacr.projects`, which in
turn assembles :func:`spacr.data_manager.scan_project` (the size and the
unaccounted-for bytes), :func:`spacr.ports.declared_outputs` (the stage),
:meth:`spacr.artifacts.Registry.is_stale` (what is out of date) and
:func:`spacr.chaining.next_steps` (what could run next). A browser with its
own opinion about any of those is a browser that disagrees with the screen
the user opens next.
Two things about it are load-bearing rather than cosmetic.
**A project the registry has never seen is still listed.** Projects are found
by walking the disk, so a folder copied from a colleague appears the moment
the browser is pointed at its parent — with its stage, its size and the date
its files were last written. What it does *not* show is "0 stale", which
would read as *clean*: with no provenance there is nothing to compare
against, so the state column says "unknown — nothing recorded" and the note
says why.
**The scan never blocks the window.** Walking a plate folder is tens of
thousands of ``stat`` calls, and doing it on the GUI thread is a frozen
application for however long the filesystem takes. Everything goes through
:class:`spacr.qt.job_runner.JobRunner`, whose completion handlers here are
**bound methods** — read that module's docstring for what a closure connected
to ``thread.finished`` costs. The background activity spinner follows the run
registry on its own, so nothing here has to drive it.
**And neither does opening the screen, nor opening its folder chooser.**
Two things here take a path the user typed on some *other* screen: the search
folders :func:`make_project_browser_screen` seeds from the recent-source list,
and the folder the "Add folder…" dialog starts in. Both used to be settled
with an ``os.path.isdir`` on the GUI thread. Measured on one workstation, one
such folder was under an ``autofs`` mount whose
share was asleep and a single ``isdir`` on it had not returned after TWENTY
SECONDS — which is not a slow screen, it is the whole application frozen with
no traceback, and it was reported as "opening the project browser crashes
spacr". Both now ask :mod:`spacr.qt.path_probe`, which answers from a cache
and stats in the background. Those two are the whole inventory: the walk
runs on a worker, and every column, note and detail line the screen draws is
rendered from the frozen :class:`spacr.projects.ProjectSummary` that walk
already returned, so filling the table and the detail pane reads nothing.
:func:`register` is **not** called at import; read its docstring.
"""
from __future__ import annotations
import logging
import os
from typing import List, Optional, Tuple
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QAbstractItemView, QFileDialog, QHBoxLayout, QHeaderView, QLabel,
QListWidget, QListWidgetItem, QPlainTextEdit, QPushButton, QSpinBox,
QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget,
)
from ..job_runner import JobRunner
from .. import path_probe
from ..theme import SPACING, mark_surface
from .app_screen import ModuleHeader
from ..widgets.collapsible_splitter import CollapsibleSplitter
from ..widgets.sortable_table import install_sorting, table_item
from ..app_catalog import declared_app, register_declared
LOG = logging.getLogger("spacr.qt.screens.project_browser")
__all__ = ["ProjectBrowserScreen", "make_project_browser_screen", "register",
"APP_KEY", "APP_NAME", "APP_DESCRIPTION", "APP_INTRO",
"APP_CLI_NOTE", "APP_NAME_TRANSLATIONS", "COLUMNS", "MAX_PROJECTS"]
#: The registry key. Chosen once and never renamed — saved user state, the
#: command palette and ``spacr-qt project_browser`` all key off it.
APP_KEY = "project_browser"
#: The table, left to right. "State" is staleness and "Note" is the one thing
#: about the project worth saying; both come from :class:`ProjectSummary`
#: properties rather than being assembled here.
COLUMNS = ("Project", "Stage", "Size", "Files", "Last run", "State", "Note")
#: How many projects one scan will list. A user who points the browser at
#: their home directory must get a table rather than a filesystem walk.
MAX_PROJECTS = 300
#: How deep below a chosen folder to look, and the range the spin box offers.
#: The default is the shape people have: a folder of experiments, each
#: holding plates.
DEFAULT_DEPTH = 2
MAX_DEPTH = 6
[docs]
class ProjectBrowserScreen(QWidget):
"""The browser: a roots picker, a table of projects, and a detail pane.
:param threaded: ``False`` runs the scan inline through the same
:class:`~spacr.qt.job_runner.JobRunner` code path, so a test drives
the real thing synchronously.
:param roots: folders to search on the first scan.
:param parent: parent widget; ownership only.
"""
#: A scan finished; carries how many projects it listed.
scanned = Signal(int)
#: Something went wrong, in one line fit for a status bar.
failed = Signal(str)
#: The user chose a project. Carries its absolute root, for a host that
#: wants to seed another screen with it.
project_chosen = Signal(str)
def __init__(self, parent: Optional[QWidget] = None, *,
threaded: bool = True,
roots: Tuple[str, ...] = ()) -> None:
"""Build the project browser.
The project table ("Projects") and the detail pane
("Project details") fold by their headings and share a draggable
edge in a
:class:`~spacr.qt.widgets.collapsible_splitter.CollapsibleSplitter`.
:param parent: parent widget, or ``None``.
:param threaded: scan on a worker thread. Set ``False`` in tests so a
refresh finishes before it returns.
:param roots: folders to scan for projects.
"""
super().__init__(parent)
self.setObjectName("ProjectBrowser")
self._summaries: Tuple = ()
self._roots: List[str] = [str(r) for r in roots if r]
self._jobs = JobRunner(self, threaded=threaded, app_key=APP_KEY)
self._jobs.job_failed.connect(self._on_job_failed)
outer = QVBoxLayout(self)
outer.setContentsMargins(SPACING["md"], SPACING["md"],
SPACING["md"], SPACING["md"])
outer.setSpacing(SPACING["sm"])
header = ModuleHeader(
APP_NAME,
description=APP_DESCRIPTION,
instruction="Add the folder your projects live in, then pick "
"one to see what it holds.",
)
self._header = header
outer.addWidget(header)
controls = QHBoxLayout()
controls.setSpacing(SPACING["sm"])
self._add = QPushButton("Add folder…")
self._add.setToolTip("Search a folder for spaCR projects")
self._add.clicked.connect(self.choose_root)
controls.addWidget(self._add)
self._forget = QPushButton("Remove")
self._forget.setToolTip("Stop searching the selected folder")
self._forget.clicked.connect(self.forget_selected_root)
controls.addWidget(self._forget)
controls.addWidget(QLabel("Depth"))
self._depth = QSpinBox()
self._depth.setRange(0, MAX_DEPTH)
self._depth.setValue(DEFAULT_DEPTH)
self._depth.setToolTip(
"How many folder levels below each search folder to look. "
"Descent stops at a project, so a project's merged/ is never "
"listed as a project of its own.")
controls.addWidget(self._depth)
self._rescan = QPushButton("Scan")
self._rescan.setToolTip("Walk the search folders again")
self._rescan.clicked.connect(self.rescan)
controls.addWidget(self._rescan)
from ..widgets.measurements_example import install_test_data_button
example = install_test_data_button(
self, controls,
lambda folder, _db: self.add_root(os.path.dirname(str(folder))),
say=lambda text: self._status.setText(text))
example.setObjectName("ProjectBrowserTestDataButton")
controls.addStretch(1)
self._status = QLabel("Add a folder to search for projects.")
self._status.setObjectName("ProjectBrowserStatus")
self._status.setWordWrap(True)
controls.addWidget(self._status, 2)
outer.addLayout(controls)
self._root_list = QListWidget()
self._root_list.setObjectName("ProjectBrowserRoots")
self._root_list.setMaximumHeight(70)
self._root_list.setToolTip("Folders searched for projects")
outer.addWidget(self._root_list)
split = CollapsibleSplitter(Qt.Horizontal,
persist_key="project_browser::body")
self._table = QTableWidget(0, len(COLUMNS))
install_sorting(self._table)
self._table.setHorizontalHeaderLabels(list(COLUMNS))
self._table.setSelectionBehavior(QAbstractItemView.SelectRows)
self._table.setSelectionMode(QAbstractItemView.SingleSelection)
self._table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self._table.setSortingEnabled(True)
self._table.verticalHeader().setVisible(False)
header = self._table.horizontalHeader()
header.setStretchLastSection(True)
header.setSectionResizeMode(QHeaderView.ResizeToContents)
self._table.itemSelectionChanged.connect(self._on_selection_changed)
self._table.itemDoubleClicked.connect(self._on_double_clicked)
split.add_section(self._table, "Projects",
persist_key="project_browser/Projects", stretch=3)
self._detail = QPlainTextEdit()
self._detail.setObjectName("ProjectBrowserDetail")
self._detail.setReadOnly(True)
self._detail.setPlaceholderText(
"Pick a project to see its stages, what is stale and why, and "
"what could run next.")
split.add_section(self._detail, "Project details",
persist_key="project_browser/Project details",
stretch=2)
mark_surface(self._root_list, self._table, self._detail)
self._body_splitter = split
outer.addWidget(split, 1)
self._refresh_root_list()
if self._roots:
self.rescan()
from ..dnd import install_for
install_for(self, "project_browser")
from .settings_model import retarget_field_tooltips
retarget_field_tooltips(self)
[docs]
def roots(self) -> Tuple[str, ...]:
"""The folders that will be searched."""
return tuple(self._roots)
[docs]
def add_root(self, path: str, *, scan: bool = True) -> bool:
"""Add a folder to search. ``True`` when it was not already there.
:param path: folder to add; ``~`` is expanded and the path made
absolute. It is also recorded as a recent folder.
"""
path = os.path.abspath(os.path.expanduser(str(path or "")))
if not path or path in self._roots:
return False
self._roots.append(path)
path_probe.isdir(path)
self._refresh_root_list()
try:
from ..prefs import push_recent_source
push_recent_source(APP_KEY, path)
except Exception:
LOG.debug("could not record the recent folder", exc_info=True)
if scan:
self.rescan()
return True
[docs]
def choose_root(self) -> None:
"""Ask for a folder and add it.
The dialog opens on a folder the probe cache has already confirmed
rather than simply on the last one searched; :meth:`_start_directory`
says why that distinction is the difference between a chooser and a
frozen window.
"""
path = QFileDialog.getExistingDirectory(
self, "Search this folder for spaCR projects",
self._start_directory())
if path:
path = os.path.abspath(os.path.expanduser(str(path)))
path_probe.prime(path, True)
self.add_root(path)
def _start_directory(self) -> str:
"""Where the folder chooser opens. Stats nothing, ever.
NOT ``self._roots[-1]``, which is what this was, and which is the
same twenty-second freeze as the seeding one button along: Qt stats
and then LISTS the start directory before it draws the dialog, so
handing it a remembered ``/nas_mnt`` root hangs the very click that
asked for the chooser.
`path_probe.isdir` answers from its cache and says *no* to a folder it
has not probed yet. That is the pessimistic direction and the right
one here, because the two costs are not comparable: skipping a root
costs a dialog that opens at the home directory, and stating one costs
the application. It also needs no `path_probe.probes.answered`
subscription to recover -- unlike a gate in front of something the
screen PAINTS, this one leaves nothing on screen to be wrong. The
answer lands in the cache moments later and the next click uses it,
and in a real session it is there already: the factory queued a probe
for every seeded root, and :meth:`add_root` queues one for every root
added since.
The roots are tried newest-first rather than only the newest being
considered, so one root still waiting on its probe does not cost the
user the several that have already answered.
One residue is left, and it is the smallest one available:
`path_probe._stat_with_timeout` reports a stat that never came back as
``True``, so a mount that is not merely asleep but dead can still be
cached as a directory and handed to the dialog. Narrowing that further
is `path_probe`'s job, not this screen's. What this gate removes is
the far commoner case -- a remembered root nobody has asked about at
all, which is every root on the first open of every session.
The home directory is the fallback because it is where Qt was
already opening this dialog before any root had been remembered, and
it is where the user's shell and desktop have it mounted anyway. It
is not probed: a network home can be slow too, but a probe cannot
help — there is nowhere further to fall back to, and refusing to open
a chooser at all is worse than any wait.
"""
for root in reversed(self._roots):
if path_probe.isdir(root):
return root
return os.path.expanduser("~")
[docs]
def forget_selected_root(self) -> None:
"""Drop the selected search folder and scan again."""
row = self._root_list.currentRow()
if 0 <= row < len(self._roots):
self._roots.pop(row)
self._refresh_root_list()
self.rescan()
def _refresh_root_list(self) -> None:
"""Rebuild the list of scanned root folders."""
self._root_list.clear()
for path in self._roots:
self._root_list.addItem(QListWidgetItem(path))
[docs]
def rescan(self) -> None:
"""Walk the search folders again, off the GUI thread."""
roots = list(self._roots)
if not roots:
self._summaries = ()
self._table.setRowCount(0)
self._detail.setPlainText("")
self._status.setText("Add a folder to search for projects.")
self.scanned.emit(0)
return
depth = int(self._depth.value())
self._jobs.cancel()
self._rescan.setEnabled(False)
self._status.setText(
f"scanning {len(roots)} folder(s), {depth} level(s) deep…")
self._jobs.submit(lambda r=roots, d=depth: _scan(r, d),
self._on_walk_done)
def _on_walk_done(self, outcome) -> None:
"""One walk landed: show its table, or say why there is none.
GUI thread only — a bound method, always. A FAILURE ARRIVES HERE,
through the completion handler, rather than on the runner's
``job_failed`` signal; :func:`_scan` says why that matters.
"""
ok, payload = outcome
if ok:
self._on_scanned(payload)
else:
self._on_job_failed(str(payload))
def _on_scanned(self, summaries) -> None:
"""Show a finished scan. GUI thread only — a bound method, always."""
self._rescan.setEnabled(True)
self._summaries = tuple(summaries or ())
self._fill_table()
unknown = sum(1 for s in self._summaries if not s.known)
note = f", {unknown} not in the registry" if unknown else ""
self._status.setText(
f"{len(self._summaries)} project(s){note}."
if self._summaries else
"No projects found — try a parent folder, or a greater depth.")
self.scanned.emit(len(self._summaries))
def _on_job_failed(self, message: str) -> None:
"""Something failed. Say so without a modal, and re-enable the button.
Never a dialog: a browser that pops one up per unreadable folder is
unusable on a machine with a stale network mount.
The walk's own failures reach this through :meth:`_on_walk_done`. It
stays connected to ``JobRunner.job_failed`` as well, for the failures
that are not the walk's — a completion handler that raised, or trouble
in the runner itself — because those must still be said out loud.
"""
self._rescan.setEnabled(True)
LOG.info("project browser: %s", message)
self._status.setText(message)
self.failed.emit(message)
def _fill_table(self) -> None:
"""Fill the project table, one row per project found.
Sorting is switched off while rows are inserted, or the table re-orders
on every write and the next row index is no longer the row just filled.
Each row carries its project root, so a re-sorted table still selects
the project the user clicked rather than whatever is now at that index,
and the size column sorts on its byte count rather than on ``"1.2 GB"``.
"""
from ...data_manager import human_bytes
self._table.setSortingEnabled(False)
self._table.setRowCount(len(self._summaries))
for row, summary in enumerate(self._summaries):
cells = (
summary.name,
summary.stage_label,
human_bytes(summary.size_bytes),
f"{summary.n_files:,}",
(summary.last_run_utc or "never").replace("+00:00", ""),
summary.staleness_note(),
summary.note(),
)
for column, text in enumerate(cells):
item = table_item(str(text))
if column == 0:
item.setToolTip(summary.root)
item.setData(Qt.UserRole, summary.root)
elif column == 2:
item.setData(Qt.UserRole, int(summary.size_bytes))
self._table.setItem(row, column, item)
self._table.setSortingEnabled(True)
self._detail.setPlainText("")
[docs]
def summaries(self) -> Tuple:
"""Every :class:`spacr.projects.ProjectSummary` currently listed."""
return self._summaries
[docs]
def selected_root(self) -> str:
"""The selected project's root, or ``""``."""
items = self._table.selectedItems()
if not items:
return ""
item = self._table.item(items[0].row(), 0)
return str(item.data(Qt.UserRole) or "") if item is not None else ""
[docs]
def summary_for(self, root: str):
"""The listed summary for one root, or ``None``.
:param root: absolute project root, compared exactly with each
summary's ``root``.
"""
for summary in self._summaries:
if summary.root == root:
return summary
return None
def _on_selection_changed(self) -> None:
"""Show the detail for the newly selected project."""
self.show_detail(self.selected_root())
def _on_double_clicked(self, _item) -> None:
"""Announce the double-clicked project as the one chosen.
:param _item: the activated cell; the root is read off the selection, so
it is not used.
"""
root = self.selected_root()
if root:
self.project_chosen.emit(root)
[docs]
def show_detail(self, root: str) -> str:
"""Draw the detail pane for one project. Returns what it drew.
:param root: absolute project root as listed; a root with no listed
summary clears the pane and returns ``""``.
"""
summary = self.summary_for(root)
if summary is None:
self._detail.setPlainText("")
return ""
from ...projects import format_project
lines = [format_project(summary), "", "Stages"]
for state in summary.modules:
lines.append(f" {state.describe()}")
if summary.stale or summary.missing:
lines.append("")
lines.append("Out of date")
for entry in summary.stale:
lines.append(f" {entry.describe()}")
explanation = entry.explain()
if explanation:
lines.append(f" {explanation}")
for entry in summary.missing:
lines.append(f" {entry.describe()}")
elif not summary.staleness_known:
lines.append("")
lines.append(
"Nothing here has a run record, so no result can be checked "
"against what produced it. Run a module on this project and "
"spaCR starts keeping one.")
if summary.next_steps:
lines.append("")
lines.append("What could run next")
for module, blocked in summary.next_steps:
lines.append(f" {module}"
+ (f" — blocked: {blocked}" if blocked else ""))
text = "\n".join(lines)
self._detail.setPlainText(text)
return text
[docs]
def is_busy(self) -> bool:
"""True while a scan is in flight."""
return self._jobs.is_busy()
[docs]
def active_jobs(self) -> int:
"""How many worker threads are still winding down."""
return self._jobs.active_jobs()
[docs]
def closeEvent(self, event): # noqa: N802 - Qt name
"""Stop background work and unlink before going away.
:param event: the Qt close event.
"""
self._jobs.shutdown()
super().closeEvent(event)
def _browse(roots: List[str], depth: int):
"""Do the walk. Runs on a worker thread and touches no widget.
Imported here rather than at module scope so opening any other screen
does not pay for :mod:`spacr.projects` and the registry machinery under
it.
"""
from ...projects import browse
return browse(roots, depth=depth, limit=MAX_PROJECTS)
def _scan(roots: List[str], depth: int):
"""Walk, and hand a failure back as a value instead of raising it.
Runs on a worker thread and touches no widget.
:returns: ``(True, summaries)`` when the walk finished, ``(False,
message)`` when it did not.
WHY IT DOES NOT SIMPLY RAISE. ``JobRunner.job_failed`` is **not**
generation-guarded: :meth:`~spacr.qt.job_runner.JobRunner.cancel` drops
the *results* of the jobs it abandons, and their failures are emitted
anyway. Two scans in quick succession — one extra click on "Scan", or a
second folder dropped on the table — could therefore have the first
walk's error land after the second walk's finished table, replace the
project count with a stale message about a folder no longer being
searched, and re-enable the Scan button underneath a walk still in
flight. A failure returned as a *value* travels back through ``on_done``,
which is generation-guarded, so a superseded walk's failure is discarded
with everything else the cancel abandoned.
The message is spelled the way the threaded path already spelled it —
``TypeName: text``, which is the last line of the traceback
``JobRunner._on_worker_error_text`` used to hand over — so the status bar
of a real, threaded browser reads exactly as it did before. The
``threaded=False`` path a test drives synchronously used to get the bare
``str(exc)`` from ``JobRunner._fail`` and now gets the same prefixed line
as everything else; the two spellings agreeing is the point of putting
the wording here rather than in two places.
"""
try:
return True, _browse(roots, depth)
except Exception as exc: # noqa: BLE001
detail = str(exc).strip()
return False, (f"{type(exc).__name__}: {detail}" if detail
else type(exc).__name__)
[docs]
def make_project_browser_screen(app_key: Optional[str] = None) -> QWidget:
"""Factory handed to :func:`spacr.qt.app.register_app`.
Seeds the search folders from the ones the user last pointed any
source-taking screen at, so the browser is useful on first open instead
of empty.
"""
roots: Tuple[str, ...] = ()
try:
from ..prefs import get_recent_sources
remembered = [os.path.dirname(p.rstrip(os.sep)) or p
for p in get_recent_sources(APP_KEY, limit=4)]
remembered += get_recent_sources("mask", limit=2)
roots = tuple(dict.fromkeys(
p for p in remembered
if p and path_probe.exists(p, default=True, want_dir=True)))
except Exception:
LOG.debug("could not read the recent folders", exc_info=True)
return ProjectBrowserScreen(roots=roots)
_ROW = declared_app(APP_KEY)
APP_NAME = _ROW.name
APP_DESCRIPTION = _ROW.desc
APP_INTRO = _ROW.intro
APP_CLI_NOTE = _ROW.cli_note
APP_NAME_TRANSLATIONS = _ROW.translations
[docs]
def register() -> bool:
"""Put the Project Browser in the app registry. Idempotent.
Called from :data:`spacr.qt.SELF_REGISTERING_MODULES`, which
:func:`spacr.qt.run` runs after ``spacr.qt.app`` is fully executed and
before ``MainWindow.__init__`` reads the registry.
The row itself -- the key, the name, the blurb, the section, the "no
headless run" sentence, the API doc link and the nine translations of the
display name -- is declared in :mod:`spacr.qt.app_catalog`.
:func:`spacr.qt.app.register_app` distributes those into the four tables
each used to need a hand-edit in, and this function's whole job is to name
which row. That is what lets the app be registered without importing this
module at all: the launch reads the table, and the screen is imported when
somebody opens it.
:returns: ``True`` if this call is what registered it. Safe to call
again: a module imported twice, or a test that re-imports it, must
not raise on the duplicate key.
"""
return register_declared(__name__) is not None