Source code for spacr.qt.widgets.file_list
"""Edit ordered settings that accept one or more file paths.
:class:`FilePathListWidget` appends files from a multi-select dialog or by
dropping files and folders. A dropped folder contributes sorted matching files
from one directory level. Duplicate paths are rejected, missing paths are
marked before execution, and entries can be reordered when plate order matters.
The widget reads and writes a plain ``list[str]`` so Qt settings, settings CSVs,
and command-line calls share the same value representation.
"""
from __future__ import annotations
import logging
import os
import re
from typing import Any, Iterable, List, Sequence
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import (QDragEnterEvent, QDragMoveEvent, QDropEvent,
QFontMetrics)
from PySide6.QtWidgets import (
QAbstractItemView,
QFileDialog,
QHBoxLayout,
QLabel,
QLineEdit,
QListWidget,
QListWidgetItem,
QPushButton,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
from .. import path_probe
from ..i18n import tr
from .sortable_table import install_sorting, table_item
LOG = logging.getLogger(__name__)
def _pair_tokens(path: str) -> set[str]:
"""Extract the tokens a filename can be paired on.
Zero padding in ``plate 007`` is normalised so it pairs with ``plate7``,
and words that appear in every file -- ``score``, ``count``, ``results``
and the model names -- are dropped, because a token every candidate
shares pairs nothing with anything.
:param path: the file.
:returns: its distinguishing tokens, lowercased.
"""
stem = os.path.splitext(os.path.basename(os.fspath(path)))[0].casefold()
stem = re.sub(r"plate[\s_-]*0*(\d+)", r"plate\1", stem)
generic = {"score", "scores", "count", "counts", "result", "results",
"unique", "combinations", "csv", "maxvit", "xgb"}
return {token for token in re.findall(r"[a-z]+\d+|\d+|[a-z]+", stem)
if token not in generic and len(token) > 1}
#: What a measurements DATABASE is called, by extension.
#:
#: Databases are told apart from tables by extension and never by header.
#: :func:`side_for_header` opens a file as text, and a sqlite file read as
#: CSV yields a header carrying neither a gRNA nor a count -- which scores it
#: as a SCORE table. A binary database silently filed as the regression's
#: response is exactly the failure this column exists to remove.
DATABASE_EXTENSIONS = (".db", ".sqlite", ".sqlite3")
_DATABASE_GENERIC = frozenset({
"measurements", "measurement", "database", "sqlite", "data", "merged",
"results", "result", "analysis"})
[docs]
def is_database_path(path) -> bool:
"""``True`` when ``path`` names a measurements database by extension.
One rule, in one place, for the widget that files a dropped path into a
column and for the drop handler that catches the same file when it lands
on the screen around the widget. Two copies would disagree the first time
somebody's database was called ``plate1.sqlite``.
:param path: file path as a string or path-like object; its extension is
compared, case-insensitively, with ``DATABASE_EXTENSIONS``.
"""
return os.path.splitext(os.fspath(path))[1].lower() in DATABASE_EXTENSIONS
def _database_tokens(path, *, depth: int = 2) -> set[str]:
"""Pairing tokens for a database, its parent folders included.
``_pair_tokens`` reads the basename, which is all a CSV has. A database
has less: spaCR writes every plate's to ``<plate>/measurements/
measurements.db``, so basename tokens are identical for every plate,
every candidate ties, and the tie rule leaves them all unpaired. The
plate is named by the FOLDER, exactly as ``multi_database._label_for``
already assumes when it disambiguates two databases for a legend.
"""
full = os.fspath(path)
tokens = _pair_tokens(full) - _DATABASE_GENERIC
parent = os.path.dirname(full)
for _ in range(max(0, int(depth))):
name = os.path.basename(parent)
if not name:
break
folder = _pair_tokens(name) - _DATABASE_GENERIC
tokens |= folder
parent = os.path.dirname(parent)
if folder:
break
return tokens
def _plate_label(tokens) -> str:
"""The longest ``plate...`` token, which is what names the row."""
plates = sorted((token for token in tokens if token.startswith("plate")),
key=len, reverse=True)
return plates[0] if plates else ""
def _best_unique(left: set, tokens: Sequence[set], unused: set):
"""Return the unique best token-match index in ``unused``.
A candidate must overlap ``left`` and strictly beat the runner-up. Return
``None`` for no overlap or a tie so ambiguous rows remain unpaired.
"""
ranked = sorted(((len(left & tokens[index]), index) for index in unused),
reverse=True)
if ranked and ranked[0][0] > 0 and (
len(ranked) == 1 or ranked[0][0] > ranked[1][0]):
return ranked[0][1]
return None
[docs]
def suggest_file_pairs(scores: Sequence[str], counts: Sequence[str], *,
databases: Sequence[str] = ()) -> list[dict]:
"""Propose visible score/count/database rows by filename tokens.
A proposal is never authoritative: the editable table is the contract the
user confirms. Unique best matches are used; ties remain unpaired.
``databases`` is keyword-only and defaults to nothing, so the two-argument
call every existing caller makes still means what it did. A database is
matched against BOTH cells of a row it may join -- a plate is named by its
score CSV as often as by its count CSV -- and one that matches nothing is
listed on its own row rather than being dropped or guessed onto row 0.
:param scores: score CSV paths; each starts a row, in the order given.
:param counts: count CSV paths; each is paired with the score whose
filename tokens it uniquely matches best, unmatched ones fill score
rows still missing a count, and any left over get rows of their own.
"""
unused = set(range(len(counts)))
count_tokens = [_pair_tokens(path) for path in counts]
rows = []
for score in scores:
left = _pair_tokens(score)
match = _best_unique(left, count_tokens, unused)
if match is not None:
unused.remove(match)
common = left & (count_tokens[match] if match is not None else set())
rows.append({"plate": _plate_label(common),
"score": os.fspath(score),
"count": os.fspath(counts[match]) if match is not None
else None,
"database": None})
leftover = iter(sorted(unused))
for row in rows:
if row["count"] is not None:
continue
index = next(leftover, None)
if index is None:
break
unused.discard(index)
row["count"] = os.fspath(counts[index])
for index in sorted(unused):
rows.append({"plate": "", "score": None,
"count": os.fspath(counts[index]), "database": None})
return _number_unlabelled_plates(_attach_databases(rows, databases))
def _number_unlabelled_plates(rows: list[dict]) -> list[dict]:
"""Give every row without a parsed plate name one generated from its order.
`plate 1`, `plate 2`, in row order. The rule is that a name shared by the
files is used when there is one, and a row whose files carry no name gets
one generated from its position instead.
NUMBERED BY ROW, NOT BY A COUNTER OVER THE UNLABELLED. A table whose
second row is parsed as `plate7` would otherwise run `plate 1`, `plate7`,
`plate 2`, which reads as though the middle row were out of sequence. The
number is the row's position and nothing else, so a generated label says
where the row is rather than how many blank ones preceded it.
REGENERATED ON EVERY PROPOSAL, which is what makes it safe. A generated
label is a default the user is expected to overwrite, and `_repropose`
rebuilds every row from scratch -- so a row labelled `plate 2` that ends
up first after a deletion is renumbered rather than left asserting a fact
about a plate. A label the user has actually typed is pinned by the
widget and survives this, exactly as a manually placed database does.
:param rows: proposed rows, in the order they will be shown.
:returns: the same rows, with blank plate cells filled in.
"""
for position, row in enumerate(rows, start=1):
if row.get("score") and row.get("count") and not row.get("plate"):
row["plate"] = f"plate {position}"
return rows
def _attach_databases(rows: list[dict], databases: Sequence[str]) -> list[dict]:
"""Fill each row's ``database`` cell by token, appending what is left over."""
paths = [os.fspath(path) for path in (databases or [])]
tokens = [_database_tokens(path) for path in paths]
unused = set(range(len(paths)))
for row in rows:
left = set()
for side in ("score", "count"):
if row.get(side):
left |= _pair_tokens(row[side])
if row.get("plate"):
left.add(row["plate"])
match = _best_unique(left, tokens, unused) if unused else None
if match is None:
row.setdefault("database", None)
continue
unused.remove(match)
row["database"] = paths[match]
if not row.get("plate"):
row["plate"] = _plate_label(left & tokens[match])
for index in sorted(unused):
rows.append({"plate": _plate_label(tokens[index]), "score": None,
"count": None, "database": paths[index]})
return rows
[docs]
def side_for_header(path) -> str:
"""``'count'`` when the file's header names a gRNA and a count, else
``'score'``.
Read from the header rather than the filename: a count export carries a
gRNA name and a count, a score export carries neither, and that is true
whatever the file is called.
Module-level because two screens ask the same question of the same files.
Regression asks it through :class:`PairedFileTableWidget`; Parameter Sweep
holds its two sides in separate list widgets and asks it through
``spacr.qt.dnd_handlers.SweepInputsDropHandler``. A second copy of this
rule would drift, and the direction it would drift in is silent: a count
table filed as a score is not an error, it is a wrong regression.
:param path: CSV file whose first row is read as the header; a file that
cannot be read counts as a score file.
"""
import csv as _csv
try:
with open(path, newline="", encoding="utf-8",
errors="replace") as handle:
header = {str(name).strip().lower()
for name in next(_csv.reader(handle), [])}
except (OSError, _csv.Error):
return "score"
return ("count" if {"grna", "grna_name"} & header and "count" in header
else "score")
[docs]
class PairedFileTableWidget(QWidget):
"""Editable one-row-per-plate score/count input and measurement link.
A row is ONE PLATE: its score and count source paths, optional named
tables within those sources, and the separate measurements database
that plate's per-object tables live in. The source paths are
filled BY ADDITION -- every arrival re-proposes the whole table from
filename tokens -- so databases dropped in the opposite order to the files
still land on the right plates. That is the rule, and the
third column obeys it rather than keeping a list of its own.
A plate with NO database is legal and is not an error: the regression is
fitted on scores and counts. The database is what makes that plate's
measurements available downstream, so its absence disables the plate
there instead of failing the run.
:param value: the table already saved, one entry per plate. Everything
that arrives afterwards RE-PROPOSES the whole table from filename
tokens rather than appending, which is what lets databases dropped in
the opposite order to the source files still land on the right plates.
:param parent: parent widget.
"""
value_changed = Signal()
_EMPTY_STATUS = ("Drop score and count CSVs here. Drop a measurements "
"database on a plate row to attach it to that plate.")
def __init__(self, value=None, parent=None):
"""Build the one-row-per-plate table.
:param value: the rows already saved.
:param parent: parent widget.
"""
super().__init__(parent)
self._scores: list[str] = []
self._counts: list[str] = []
self._databases: list[str] = []
self._pinned: dict[str, dict] = {}
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.table = QTableWidget(0, 7, self)
install_sorting(self.table)
self.table.setHorizontalHeaderLabels(
["Plate / proposal", tr("Score source"), tr("Count source"),
"Measurements DB", "Plate rule", tr("Score table"),
tr("Count table")])
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.itemChanged.connect(lambda *_: self.value_changed.emit())
self._fit_columns_to_their_headers()
layout.addWidget(self.table)
self.status = QLabel(self._EMPTY_STATUS, self)
self.status.setWordWrap(True)
self.status.setProperty("role", "hint")
layout.addWidget(self.status)
buttons = QHBoxLayout()
add_scores = QPushButton(tr("Add score tables…"), self)
add_counts = QPushButton(tr("Add count tables…"), self)
add_databases = QPushButton("Add measurements DBs…", self)
add_score_stores = QPushButton(tr("Add score store folder…"), self)
add_count_stores = QPushButton(tr("Add count store folder…"), self)
add_score_stores.setObjectName("ALPHA_FEATURES576")
add_count_stores.setObjectName("ALPHA_FEATURES576")
add_row = QPushButton("Add empty pair", self)
up = QPushButton("↑", self)
down = QPushButton("↓", self)
remove = QPushButton("Remove", self)
add_scores.clicked.connect(lambda: self._pick("score"))
add_counts.clicked.connect(lambda: self._pick("count"))
add_databases.clicked.connect(lambda: self._pick("database"))
add_score_stores.clicked.connect(
lambda: self._pick_store_folder("score"))
add_count_stores.clicked.connect(
lambda: self._pick_store_folder("count"))
add_row.clicked.connect(lambda: self._append_row({}))
up.clicked.connect(lambda: self._move(-1))
down.clicked.connect(lambda: self._move(1))
remove.clicked.connect(self._remove)
for button in (add_scores, add_counts, add_databases,
add_score_stores, add_count_stores, add_row, up, down,
remove):
buttons.addWidget(button)
buttons.addStretch(1)
layout.addLayout(buttons)
self.setAcceptDrops(True)
self.table.setAcceptDrops(False)
self.set_value(value)
def _refresh_alpha_visibility(self) -> None:
"""Apply the preference to named-table controls and their columns."""
from ..preferences import _apply_alpha_widgets, _is_alpha_visible
_apply_alpha_widgets(self)
visible = _is_alpha_visible("widgets", "ALPHA_FEATURES576")
for column in (self.SCORE_TABLE_COLUMN, self.COUNT_TABLE_COLUMN):
self.table.setColumnHidden(column, not visible)
#: Column index of each side in the table, so a drop lands where the user
#: aimed it rather than in whichever input the router reached first.
SIDE_COLUMNS = {"score": 1, "count": 2, "database": 3}
RULE_COLUMN = 4
SCORE_TABLE_COLUMN = 5
COUNT_TABLE_COLUMN = 6
#: The Download buttons that sit above this table, and the column each one
#: fills. The second half of the same request: align each button to its respective columns below in the table that would be perfect."
#:
#: FOUND BY THE ATTRIBUTE THE SCREEN KEEPS THEM ON, not by what a button
#: says. `AppScreen.load_the_screen_data` renames a button to "Fetching…"
#: while its download runs and `load_the_example_screen` renames it to
#: "Fetching N file(s)…", so a strip identified by button text would lose
#: its buttons in the middle of the download the user is watching.
#:
#: `None` is a button whose download fills a SETTING rather than a column:
#: Image crops writes `src`, which is a row of the same form rather than
#: anything in this table, so it has no column to sit over and trails the
#: aligned ones instead of being centred on a column it does not fill.
DOWNLOAD_BUTTONS = (
("_example_scores_button", SIDE_COLUMNS["score"]),
("_example_counts_button", SIDE_COLUMNS["count"]),
("_screen_feature_button", SIDE_COLUMNS["database"]),
("_screen_crops_button", None),
)
def _fit_columns_to_their_headers(self) -> None:
"""Widen any column too narrow to show its own heading.
MEASURED AT THE DEFAULT WIDTH, which is where a user meets this
table: Qt gives every section 100 px, and at 100 px "Plate /
proposal" draws as "late / propos" and "Measurements DB" loses its
last word. Two of five headings, unreadable before anybody has
touched anything -- and the rule is that every piece of text fits
its container.
A MINIMUM AND NOT A FIXED WIDTH. The section stays interactively
resizable, because a user who wants a narrow column is allowed to
have one; this only moves the width it STARTS at.
"""
header = self.table.horizontalHeader()
metrics = QFontMetrics(header.font())
for column in range(self.table.columnCount()):
item = self.table.horizontalHeaderItem(column)
text = item.text() if item is not None else ""
needed = metrics.horizontalAdvance(text) + 18
if header.sectionSize(column) < needed:
header.resizeSection(column, needed)
#: Sentinels for the remembered Download column widths. A column is
#: either unseen, at a width this widget set (or found acceptable), or
#: the user's -- and the third is a one-way door.
_NEVER_SET = object()
_THE_USERS = object()
def _widen_columns_for(self, columns) -> None:
"""Make sure each Download button's column can hold the button.
THE BUTTON AND THE HEADING ARE TWO NAMES FOR ONE COLUMN, so the
wider of them is what the column has to fit. Aligning a button to a
column NARROWER than the button leaves two bad choices -- overlap
the neighbour, or clip the caption -- and the strip's layout takes
the second, so "Measurements (.db)" lost its ending at the default
100 px. Widening the column removes the choice.
Done from the TABLE, which owns its header, rather than from the
strip: a layout that wrote back to the geometry it reads would
re-enter itself on the resize it caused.
Once only, and never against the user: a column the user has since
dragged narrower is theirs, and this must not drag it back on every
show.
:param columns: ``(button, column)`` pairs, column None for a button
that fills a setting rather than a column.
"""
header = self.table.horizontalHeader()
applied = getattr(self, "_download_widths", None)
if applied is None:
applied = self._download_widths = {}
self._download_widths_applied = True
for button, column in columns:
if column is None:
continue
wanted = button.sizeHint().width()
current = header.sectionSize(column)
ours = applied.get(column, self._NEVER_SET)
if ours is self._THE_USERS:
continue
if ours is not self._NEVER_SET and current != ours:
applied[column] = self._THE_USERS
continue
if current < wanted:
header.resizeSection(column, wanted)
applied[column] = header.sectionSize(column)
elif ours is self._NEVER_SET:
applied[column] = current
[docs]
def showEvent(self, event): # noqa: N802
"""Take the Download row above over this table's columns.
WHY THE TABLE REACHES UP FOR THE STRIP RATHER THAN THE OTHER WAY
ROUND. The strip is built by the generic `AppScreen`, which knows
nothing about this table's columns -- it is four buttons in a
QHBoxLayout, and every module's example-data row is built the same
way. This widget is the only object that knows what the columns ARE,
and it is the only widget in the form that is Regression's alone, so
the knowledge and the alignment are kept in the same place.
ON SHOW rather than in `__init__`: the strip is added to the section
AFTER the settings rows are, so at construction there is nothing to
find, and the widget has no parent chain to walk up either. Once
installed the layout follows the header on its own, and the call is
idempotent, so repeated shows cost one dictionary lookup.
:param event: the show event; it is not inspected, only passed on to
the base class first.
"""
super().showEvent(event)
self._refresh_alpha_visibility()
try:
self.align_download_buttons()
except Exception: # noqa: BLE001
LOG.debug("could not align the Download row", exc_info=True)
[docs]
def align_download_buttons(self) -> bool:
"""Put the Download buttons over the columns they fill.
:returns: True when the strip was found and is now following this
table's header; False when this table was not built into a screen
that has a Download row -- which is every use of it outside
Regression's Input Tables, and every test that builds it bare.
"""
screen = self._owning_screen()
if screen is None:
return False
strip = None
columns = []
for name, column in self.DOWNLOAD_BUTTONS:
button = getattr(screen, name, None)
if button is None:
return False
if strip is None:
strip = button.parentWidget()
elif button.parentWidget() is not strip:
return False
columns.append((button, column))
if strip is None:
return False
from .column_aligned_row import align_row_to_columns
self._widen_columns_for(columns)
self._download_columns = list(columns)
return align_row_to_columns(
strip, self.table.horizontalHeader(), columns,
on_invalidate=self._refit_download_columns) is not None
def _refit_download_columns(self) -> None:
"""Re-fit the Download columns to captions that have since changed.
Called by the strip's layout when a managed button's size hint
moves, which is what a translated caption does. Cheap -- four size
hints and at most four section resizes -- and guarded against
re-entry, because resizing a section makes the header emit and the
row re-lay itself.
"""
if getattr(self, "_refitting", False):
return
columns = getattr(self, "_download_columns", None)
if not columns:
return
self._refitting = True
try:
self._widen_columns_for(columns)
except Exception: # noqa: BLE001
LOG.debug("could not re-fit the Download columns", exc_info=True)
finally:
self._refitting = False
def _owning_screen(self):
"""The screen holding the Download buttons, or None.
Walked up the parent chain and recognised by the attribute the
buttons are kept on rather than by class: importing `AppScreen` here
would be a widget importing the screen that hosts it, and this file
is imported while that module is still being built.
"""
node = self.parentWidget()
while node is not None:
if hasattr(node, self.DOWNLOAD_BUTTONS[0][0]):
return node
node = node.parentWidget()
return None
[docs]
def add_paths_for_side(self, paths, side: str = "score") -> int:
"""Add files to one column of the table and re-propose the pairing.
The drop router calls this method after identifying a file's role from
its header, so a count table dropped anywhere on the widget still
fills the count column.
``'database'`` is the third side. It goes through this same method,
and therefore through the same whole-table re-proposal, rather than
through an adder of its own that would keep a private list and pair
by the order things arrived.
:param paths: File paths to add.
:param side: Target column: ``'score'``, ``'count'``, or
``'database'``.
:returns: Number of previously absent files added.
"""
if side not in self.SIDE_COLUMNS:
raise ValueError(
f"side must be 'score', 'count' or 'database'; got {side!r}.")
incoming = [os.fspath(p) for p in (paths or [])]
if not incoming:
return 0
self._rebuild_sides()
target = {"score": self._scores, "count": self._counts,
"database": self._databases}[side]
added = 0
for path in incoming:
if path not in target:
target.append(path)
added += 1
if added:
self._repropose()
self.value_changed.emit()
return added
def _rebuild_sides(self) -> None:
"""Refill the three flat lists from the table, which is the state."""
current = self.get_value()
self._scores = list(dict.fromkeys(
row["score"] for row in current if row.get("score")))
self._counts = list(dict.fromkeys(
row["count"] for row in current if row.get("count")))
self._databases = list(dict.fromkeys(
row["database"] for row in current if row.get("database")))
def _repropose(self, typed: dict | None = None) -> None:
"""Rebuild every row from tokens, then honour the manual attachments.
:param typed: the plate labels the user typed, keyed by the row's
files. Omitted, they are read from the table as it stands. A row
move or delete passes the labels it read BEFORE changing the
table, because a generated `plate 2` that a move carried to the
top no longer matches the proposal for its new position and would
otherwise be mistaken for a name the user chose.
"""
if typed is None:
typed = self._plate_labels_the_user_typed()
previous = self.get_value()
rows = suggest_file_pairs(self._scores, self._counts,
databases=self._databases)
qualified = [row for row in previous
if row.get("score_table") or row.get("count_table")]
if qualified:
restored = []
used = set()
for row in rows:
matches = [(index, prior)
for index, prior in enumerate(qualified)
if index not in used
and prior.get("score") == row.get("score")
and prior.get("count") == row.get("count")]
if not matches:
partial = [(index, prior)
for index, prior in enumerate(qualified)
if index not in used and (
prior.get("score") == row.get("score")
and not prior.get("count") or
prior.get("count") == row.get("count")
and not prior.get("score"))]
if len(partial) == 1:
matches = partial
if matches:
for index, prior in matches:
used.add(index)
keep = dict(row)
keep.update(prior)
for side in ("score", "count", "database"):
keep[side] = prior.get(side) or row.get(side)
restored.append(keep)
else:
restored.append(row)
restored.extend(prior for index, prior in enumerate(qualified)
if index not in used)
rows = restored
keys = [(row.get("score"), row.get("count")) for row in rows]
for row, key in zip(rows, keys):
label = typed.get(key) if keys.count(key) == 1 else None
if label:
row["plate"] = label
self.set_value(self._apply_pinned(rows))
def _plate_labels_the_user_typed(self) -> dict:
"""Plate labels the table holds that no proposal would have produced.
A GENERATED LABEL IS A DEFAULT AND A TYPED ONE IS A DECISION, and
`_repropose` rebuilds every row from the filenames, so without this it
discards the decision. `plate 2` on a row that has become the first
row must be renumbered; `Treated, day 3` typed by the user must not.
Told apart by RE-PROPOSING AND COMPARING rather than by a flag on the
row: whatever `suggest_file_pairs` would say for the current files is
by definition not a decision, and anything else is. That needs no
extra state to go stale, and it stays correct if the proposal rules
change -- a label that used to be generated and no longer is becomes
a decision automatically, which is the safe direction.
KEYED ON THE ROW'S FILES, NOT ITS POSITION, because position is
exactly what re-proposing is allowed to change.
:returns: ``{(score, count): label}`` for rows the user has named.
"""
proposed = suggest_file_pairs(self._scores, self._counts,
databases=self._databases)
default = {(row.get("score"), row.get("count")): row.get("plate")
for row in proposed}
typed = {}
for row in self.get_value():
key = (row.get("score"), row.get("count"))
label = row.get("plate")
if label and label != default.get(key):
typed[key] = label
return typed
def _apply_pinned(self, rows: list[dict]) -> list[dict]:
"""Move each pinned database back onto the row the user chose."""
for database, anchor in list(self._pinned.items()):
current = next((index for index, row in enumerate(rows)
if row.get("database") == database), None)
if current is None:
self._pinned.pop(database, None)
continue
target = self._row_for_anchor(rows, anchor)
if target is None or target == current:
continue
rows[current]["database"] = rows[target].get("database")
rows[target]["database"] = database
return [row for row in rows
if row.get("score") or row.get("count") or row.get("database")]
@staticmethod
def _row_for_anchor(rows: list[dict], anchor: dict):
"""The row a pin names, by its files first and its plate label last."""
for key in ("score", "count", "plate"):
value = anchor.get(key)
if not value:
continue
for index, row in enumerate(rows):
if row.get(key) == value:
return index
return None
def _anchor_for_row(self, row: int) -> dict:
"""The path a row's other cells are resolved against.
:param row: the row's position.
:returns: the anchor path, or None when the row has none yet.
"""
return {key: self._cell(row, column) or None
for key, column in (("plate", 0),
("score", self.SIDE_COLUMNS["score"]),
("count", self.SIDE_COLUMNS["count"]))}
def _cell(self, row: int, column: int) -> str:
"""One cell's text.
:param row: the row's position.
:param column: the column's position.
:returns: the text, empty when the cell is blank.
"""
editor = self.table.cellWidget(row, column)
if isinstance(editor, QLineEdit):
return editor.text().strip()
item = self.table.item(row, column)
return item.text().strip() if item else ""
def _side_for_header(self, path) -> str:
"""Which column ``path`` belongs in. See :func:`side_for_header`."""
return side_for_header(path)
def _side_for_path(self, path) -> str:
"""Which column ``path`` belongs in, databases decided by extension."""
return ("database" if is_database_path(path)
else self._side_for_header(path))
[docs]
def attach_database(self, path, row=None) -> str:
"""Attach a measurements database to one plate row, and say which.
``row`` is the row the user aimed the drop at -- an EXPLICIT
assignment, which is remembered and survives the re-proposal that the
next dropped CSV triggers.
With no row, the database is offered to the token pairing first, so
``plate2/measurements/measurements.db`` finds plate 2 wherever that
row happens to be. Only when nothing in its path names a plate does
it fall back to the first row that has no database -- and the
returned sentence, also shown under the table, NAMES that row. A
database attached to row 0 in silence is a plate's measurements
quietly credited to another plate.
Returns the sentence, so a caller with a console logs the same words
the user is reading.
:param path: path to the measurements database, as a string or
path-like object; an empty path raises :class:`ValueError`.
"""
database = os.fspath(path).strip()
if not database:
raise ValueError("attach_database needs a path to a database.")
if row is not None:
row = int(row)
if not 0 <= row < self.table.rowCount():
raise IndexError(
f"row {row} is not in a table of {self.table.rowCount()} "
"rows.")
replaced = self._place_database(row, database)
message = self._describe_row(row, database, "attached to")
if replaced:
message += f" It replaced {os.path.basename(replaced)}."
self.value_changed.emit()
self._refresh_status(message)
return message
already = self._row_of_database(database)
self.add_paths_for_side([database], "database")
index = self._row_of_database(database)
if already is not None:
message = self._describe_row(index, database, "is already on")
self._refresh_status(message)
return message
if self._cell(index, self.SIDE_COLUMNS["score"]) or \
self._cell(index, self.SIDE_COLUMNS["count"]):
message = self._describe_row(index, database, "paired by filename with")
self._refresh_status(message)
return message
target = self._first_row_without_database(exclude=index)
if target is None:
message = (f"{os.path.basename(database)} is on row {index + 1} "
"of its own: no plate row is waiting for a database. "
"It will pair with a plate when that plate's CSVs "
"arrive.")
self._refresh_status(message)
return message
self.table.blockSignals(True)
self.table.removeRow(index)
self.table.blockSignals(False)
target = self._first_row_without_database()
self._place_database(target, database)
message = self._describe_row(target, database, "attached to")
message += (" It is the first row with no database: nothing in the "
"file's path named a plate.")
self.value_changed.emit()
self._refresh_status(message)
return message
[docs]
def missing_databases(self) -> list:
"""Rows whose database is not on disk: ``(row number, plate, path)``.
Row numbers are 1-based, the way the table numbers them.
Checked as soon as the path is attached and restated in the status
line, because the alternative is a run that reads its inputs, fits
nothing for four minutes, and then fails on a path the panel could
have flagged the moment the settings were loaded. A settings file is
routinely written on one machine and run on another, which is the
case that produces this.
"""
missing = []
for index, row in enumerate(self.get_value(), start=1):
database = row.get("database")
if database and not path_probe.exists(database, wait=True):
missing.append((index, row.get("plate") or "", database))
return missing
def _row_of_database(self, database: str):
"""Which row already holds this database, if any.
:param database: the database path.
:returns: the row's position, or None.
"""
column = self.SIDE_COLUMNS["database"]
for index in range(self.table.rowCount()):
if self._cell(index, column) == database:
return index
return None
def _first_row_without_database(self, exclude=None):
"""The first row still waiting for a database.
SO A DROP FILLS A GAP rather than always appending: a user who dropped
scores first and databases second expects the second drop to complete
the rows, not to start new ones.
:param exclude: a row to skip.
:returns: the row's position, or None when every row has one.
"""
column = self.SIDE_COLUMNS["database"]
for index in range(self.table.rowCount()):
if index == exclude:
continue
if not self._cell(index, column):
return index
return None
def _place_database(self, row: int, database: str) -> str:
"""Write ``database`` into ``row`` and pin it there. Returns what it
replaced, if anything."""
column = self.SIDE_COLUMNS["database"]
replaced = self._cell(row, column)
self.table.blockSignals(True)
self.table.setItem(row, column, self._database_item(database))
self.table.blockSignals(False)
if replaced and replaced != database:
self._pinned.pop(replaced, None)
self._pinned[database] = self._anchor_for_row(row)
return replaced if replaced != database else ""
@staticmethod
def _database_item(value: str) -> QTableWidgetItem:
"""One table cell holding a database path.
:param value: the path.
:returns: the cell.
"""
item = table_item(str(value))
if value and not path_probe.exists(str(value), wait=True):
item.setForeground(Qt.red)
item.setToolTip(f"{value}\n\nThis database is not on disk right "
"now, so this plate has no measurements to join.")
return item
def _describe_row(self, row: int, database: str, verb: str) -> str:
"""One line saying what a row now holds, for the status area.
:param row: the row's position.
:param database: the database it holds.
:param verb: what just happened to it.
:returns: the description.
"""
plate = self._cell(row, 0)
label = f"{plate} (row {row + 1})" if plate else f"row {row + 1}"
return f"{os.path.basename(database)} {verb} {label}."
def _refresh_status(self, message: str = "", *,
check_paths: bool = True) -> None:
"""Put one line in the status area.
:param message: the line.
"""
rows = self.get_value()
parts = [message] if message else []
attached = [row for row in rows if row.get("database")]
if rows:
noun = "row" if len(rows) == 1 else "rows"
parts.append(f"{len(attached)} of {len(rows)} plate {noun} "
"carry a measurements database.")
if check_paths:
missing = self._missing_databases_cache = self.missing_databases()
else:
missing = getattr(self, '_missing_databases_cache', [])
if missing:
named = "; ".join(
f"{plate or f'row {number}'}: {path}"
for number, plate, path in missing)
parts.append(f"NOT ON DISK — {named}. Fix or clear these before "
"the run: they are read after it starts.")
unnamed = []
for number, row in enumerate(rows, start=1):
for side in ('score', 'count'):
path = str(row.get(side) or '').lower()
if (path.startswith(('postgresql://', 'postgres://'))
or path.endswith(('.db', '.sqlite', '.sqlite3',
'.duckdb', '.ddb', '.parquetdb'))):
if not row.get(f'{side}_table'):
unnamed.append(f'{side} {number}')
if unnamed:
parts.append(tr("Name the table for {sources} before running.",
sources=', '.join(unnamed)))
self.status.setText(" ".join(parts) or self._EMPTY_STATUS)
@staticmethod
def _dropped(event):
"""Whether a drag carries files this table can take.
:param event: the Qt drag event.
:returns: True when droppable.
"""
mime = event.mimeData()
if not mime.hasUrls():
return []
return [url.toLocalFile() for url in mime.urls() if url.isLocalFile()]
[docs]
def dragEnterEvent(self, event): # noqa: N802 - Qt name
"""Accept a drag carrying files this table can take.
:param event: the Qt drag event.
"""
if self._dropped(event):
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dragMoveEvent(self, event): # noqa: N802 - Qt name
"""Keep accepting while droppable files stay over the table.
:param event: the Qt drag event.
"""
if self._dropped(event):
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dropEvent(self, event): # noqa: N802 - Qt name
"""Route each dropped file to the column, and row, it belongs in.
A drop over the score or count column goes there regardless of what
the file looks like -- the user aimed it. A drop anywhere else is
sorted by header, so dropping the whole set at once still fills both
columns correctly.
A DATABASE is decided by its extension and never by aim: a ``.db``
dropped on the score column is a mis-aim, not a request to fit the
regression on a sqlite file, and a CSV dropped on the database column
is likewise still a CSV. What aim adds for a database is the ROW --
dropping it on a plate's row attaches it to THAT plate, which is the
one thing the token pairing cannot know when the file is called
``measurements.db`` like everybody else's.
:param event: the drop event; its local file URLs and drop position
are read, and it is ignored when it carries no usable paths.
"""
paths = self._dropped(event)
if not paths:
event.ignore()
return
position = event.position().toPoint() if hasattr(event, "position") \
else event.pos()
local = self.table.viewport().mapFrom(self, position)
inside = self.table.viewport().rect().contains(local)
column = self.table.columnAt(local.x()) if inside else -1
row = self.table.rowAt(local.y()) if inside else -1
aimed = None
for side, index in self.SIDE_COLUMNS.items():
if column == index and inside:
aimed = side
break
for path in paths:
natural = self._side_for_path(path)
if natural == "database":
self.attach_database(
path, row if aimed == "database" and row >= 0 else None)
else:
side = aimed if aimed in ("score", "count") else natural
self.add_paths_for_side([path], side)
event.acceptProposedAction()
def _pick(self, side: str) -> None:
"""Ask for a file for one side of the current row.
:param side: which column it fills.
"""
if side == "database":
title, filters = ("Add measurements databases",
"Databases (*.db *.sqlite *.sqlite3)")
else:
title, filters = (tr("Add {side} tables", side=side),
"Tables (*.csv *.tsv *.txt *.parquet *.feather "
"*.xlsx *.db *.sqlite *.sqlite3 *.duckdb "
"*.ddb);;All files (*)")
paths, _ = QFileDialog.getOpenFileNames(self, title, "", filters)
if not paths:
return
self.add_paths_for_side(paths, side)
def _pick_store_folder(self, side: str) -> None:
"""Add a Parquet store folder to the selected input side."""
path = QFileDialog.getExistingDirectory(
self, tr("Add {side} store folder", side=side))
if not path:
return
if not path.lower().endswith('.parquetdb'):
self.status.setText(tr("Choose a .parquetdb store folder."))
return
self.add_paths_for_side([path], side)
def _append_row(self, row: dict) -> None:
"""Add one row to the table.
:param row: the row's values.
"""
index = self.table.rowCount()
self.table.insertRow(index)
values = (row.get("plate") or "", row.get("score") or "",
row.get("count") or "", row.get("database") or "",
row.get("rule") or "resolved at run",
row.get("score_table") or "",
row.get("count_table") or "")
for column, value in enumerate(values):
if column in (self.SCORE_TABLE_COLUMN, self.COUNT_TABLE_COLUMN):
edit = QLineEdit(self.table)
edit.setObjectName("ALPHA_FEATURES576")
edit.setText(str(value))
edit.setPlaceholderText(tr("Table name for store"))
edit.textChanged.connect(lambda *_: self.value_changed.emit())
edit.textChanged.connect(
lambda *_: self._refresh_status(check_paths=False))
self.table.setCellWidget(index, column, edit)
continue
if column == self.SIDE_COLUMNS["database"]:
item = self._database_item(value)
else:
item = table_item(str(value))
if column == self.RULE_COLUMN:
item.setFlags(item.flags() & ~Qt.ItemIsEditable)
self.table.setItem(index, column, item)
[docs]
def set_value(self, value: Any) -> None:
"""Replace every row from a settings value.
SIGNALS BLOCKED while the rows are rebuilt: this is called when a
settings file is poured in, and one change signal per cell would
re-validate the whole form once per cell.
:param value: the rows, as the settings dict carries them.
"""
self.table.blockSignals(True)
self.table.setRowCount(0)
for row in value or []:
if isinstance(row, dict):
self._append_row(row)
self.table.blockSignals(False)
self._refresh_status()
self._refresh_alpha_visibility()
[docs]
def get_value(self) -> list[dict]:
"""Every row, in the shape the settings dict wants.
:returns: one dict per plate row.
"""
rows = []
for index in range(self.table.rowCount()):
score = self._cell(index, self.SIDE_COLUMNS["score"])
count = self._cell(index, self.SIDE_COLUMNS["count"])
database = self._cell(index, self.SIDE_COLUMNS["database"])
score_table = self._cell(index, self.SCORE_TABLE_COLUMN)
count_table = self._cell(index, self.COUNT_TABLE_COLUMN)
if score or count or database or score_table or count_table:
row = {"plate": self._cell(index, 0) or None,
"score": score or None, "count": count or None,
"database": database or None}
if score_table:
row["score_table"] = score_table
if count_table:
row["count_table"] = count_table
rows.append(row)
return rows
def _move(self, offset: int) -> None:
"""Move the selected row up or down.
:param offset: -1 for up, +1 for down.
"""
row = self.table.currentRow()
target = row + offset
if row < 0 or not 0 <= target < self.table.rowCount():
return
typed = self._labels_before_a_change()
values = self.get_value()
values[row], values[target] = values[target], values[row]
self.set_value(values)
self._repropose_after_a_change(typed)
self.table.selectRow(target)
self.value_changed.emit()
def _labels_before_a_change(self) -> dict:
"""The typed plate labels, read while the table is still unchanged."""
self._rebuild_sides()
return self._plate_labels_the_user_typed()
def _repropose_after_a_change(self, typed: dict) -> None:
"""Renumber generated plate labels the moment rows move or go.
A generated label names a row's position, so a move or a delete that
leaves it in place has it asserting a position the row no longer has
until the next file arrives.
:param typed: labels from :meth:`_labels_before_a_change`.
"""
current = self.get_value()
if any(row.get("score_table") or row.get("count_table")
for row in current):
keys = [(row.get("score"), row.get("count")) for row in current]
for position, row in enumerate(current, start=1):
key = (row.get("score"), row.get("count"))
if keys.count(key) > 1:
continue
row["plate"] = typed.get(key) or (
f"plate {position}" if row.get("score")
and row.get("count") else None)
self.set_value(current)
self._rebuild_sides()
return
self._rebuild_sides()
self._repropose(typed)
def _remove(self) -> None:
"""Remove the selected row."""
rows = sorted({index.row() for index in self.table.selectedIndexes()},
reverse=True)
typed = self._labels_before_a_change() if rows else {}
for row in rows:
self._pinned.pop(self._cell(row, self.SIDE_COLUMNS["database"]),
None)
self.table.removeRow(row)
if rows:
self._repropose_after_a_change(typed)
self._refresh_status()
self.value_changed.emit()
#: Extension groups offered in the dialog, by the kind of input a setting wants.
FILE_KIND_FILTERS: dict[str, str] = {
"table": "Tables (*.csv *.tsv *.txt *.xlsx *.parquet);;CSV (*.csv);;"
"Excel (*.xlsx);;All files (*)",
"csv": "CSV (*.csv *.tsv *.txt);;All files (*)",
"image": "Images (*.tif *.tiff *.png *.jpg *.jpeg *.bmp *.czi *.lif *.nd2);;"
"All files (*)",
"model": "Models (*.pth *.pt *.ckpt *.h5 *.joblib *.pkl);;All files (*)",
"sequencing": "Reads (*.fastq *.fq *.fastq.gz *.fq.gz);;All files (*)",
"any": "All files (*)",
}
#: Extensions a dropped *folder* contributes, per kind. A folder dropped on a
#: CSV setting must not add its PNGs.
_KIND_EXTENSIONS: dict[str, tuple[str, ...]] = {
"table": (".csv", ".tsv", ".txt", ".xlsx", ".parquet"),
"csv": (".csv", ".tsv", ".txt"),
"image": (".tif", ".tiff", ".png", ".jpg", ".jpeg", ".bmp", ".czi",
".lif", ".nd2"),
"model": (".pth", ".pt", ".ckpt", ".h5", ".joblib", ".pkl"),
"sequencing": (".fastq", ".fq", ".gz"),
"any": (),
}
[docs]
class FilePathListWidget(QWidget):
"""An ordered, de-duplicated list of input paths with picker and drop.
``single=True`` is the same control for a setting that names exactly ONE
file. It keeps the file dialog and the drop target -- which is the whole
reason these settings stopped being text boxes -- but the value it holds
and returns is a plain ``str``, choosing again REPLACES rather than
appends, and the reorder buttons are gone because one path has no order.
That distinction is not cosmetic. ``grna_csv``, ``row_csv`` and
``column_csv`` are declared ``str`` and go straight to ``pd.read_csv``,
so rendering them as a list turned a working default into
``['/path/to/barcodes_row.csv']`` the moment the screen was opened and
saved. Every run from that file was then refused by the pre-flight --
"column_csv=[...] is a list, but str is expected" -- against a value the
user had never typed and could not correct from the panel that wrote it.
:param value: what the setting already holds. A bare string stays a
string -- see above; that is the whole point of this widget.
:param kind: which file filter the chooser opens with, one of
:data:`FILE_KIND_FILTERS`. An unrecognised name falls back to
``"any"`` rather than raising, so a new setting cannot break a panel
by naming a filter that does not exist yet.
:param title: the file dialog's window title.
:param allow_folders: whether a folder may be picked. Forced off when
``single`` is set: expanding a folder into "every CSV in here" cannot
mean anything for a setting that names ONE file.
:param single: whether the setting holds one path rather than a list.
:param parent: parent widget.
"""
value_changed = Signal()
contents_changed = Signal()
def __init__(
self,
value: Any = None,
*,
kind: str = "table",
title: str = "Choose input files",
allow_folders: bool = True,
single: bool = False,
parent=None,
):
"""Build the ordered, de-duplicated path list.
:param value: the paths already saved.
:param kind: what sort of file it accepts.
:param title: the caption on its picker.
:param allow_folders: whether folders may be added.
:param single: whether only one path is allowed.
:param parent: parent widget.
"""
super().__init__(parent)
self._kind = kind if kind in FILE_KIND_FILTERS else "any"
self._title = title
self._single = bool(single)
self._allow_folders = bool(allow_folders) and not self._single
self._last_directory = ""
outer = QVBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
outer.setSpacing(4)
self._list = QListWidget(self)
self._list.setSelectionMode(QAbstractItemView.ExtendedSelection)
self._list.setAlternatingRowColors(True)
self._list.setMinimumHeight(48 if self._single else 96)
self._list.setUniformItemSizes(True)
self._list.setAcceptDrops(False)
self._list.setDragDropMode(QAbstractItemView.NoDragDrop)
outer.addWidget(self._list)
self._hint = QLabel(self._empty_hint(), self)
self._hint.setWordWrap(True)
self._hint.setProperty("role", "hint")
outer.addWidget(self._hint)
self._follow_path_probes()
self._single_line = None
if self._single:
self._list.hide()
self._hint.hide()
self._single_line = QLineEdit(self)
self._single_line.setReadOnly(True)
self._single_line.setPlaceholderText(self._empty_hint())
self._single_line.setToolTip(
"Drop one file here, or use Choose file…")
row = QHBoxLayout()
row.setSpacing(4)
if self._single_line is not None:
row.addWidget(self._single_line, 1)
self._add_files_button = QPushButton(
"Choose file…" if self._single else "Add files…", self)
self._add_files_button.setToolTip(
"Select the file this setting names. Choosing again replaces it."
if self._single else
"Select one or more files. Press again to add more from another "
"folder — each press appends to the list.")
self._add_files_button.clicked.connect(self.pick_files)
row.addWidget(self._add_files_button)
if self._allow_folders:
self._add_folder_button = QPushButton("Add folder…", self)
self._add_folder_button.setToolTip(
"Add every matching file directly inside a folder.")
self._add_folder_button.clicked.connect(self.pick_folder)
row.addWidget(self._add_folder_button)
else:
self._add_folder_button = None
if self._single:
self._up_button = self._down_button = None
else:
self._up_button = QPushButton("↑", self)
self._up_button.setToolTip(
"Move the selected file earlier in the list")
self._up_button.setMaximumWidth(
max(30, self._up_button.sizeHint().width()))
self._up_button.clicked.connect(lambda: self._move_selected(-1))
row.addWidget(self._up_button)
self._down_button = QPushButton("↓", self)
self._down_button.setToolTip(
"Move the selected file later in the list")
self._down_button.setMaximumWidth(
max(30, self._down_button.sizeHint().width()))
self._down_button.clicked.connect(lambda: self._move_selected(1))
row.addWidget(self._down_button)
self._remove_button = QPushButton("Remove", self)
self._remove_button.clicked.connect(self.remove_selected)
row.addWidget(self._remove_button)
self._clear_button = QPushButton("Clear", self)
self._clear_button.clicked.connect(self.clear)
row.addWidget(self._clear_button)
row.addStretch(1)
outer.addLayout(row)
self.setAcceptDrops(True)
self.set_value(value)
[docs]
def set_value(self, value: Any) -> None:
"""Replace the contents. Accepts None, a str, or any iterable.
A single-file widget keeps only the last of whatever it is given.
That is what loads a settings file written while these keys were
wrongly rendered as lists: ``['/x/barcodes_row.csv']`` comes back as
``/x/barcodes_row.csv`` rather than carrying the wrong shape forward.
:param value: None, one path as a string or path-like object, or an
iterable of paths; surrounding quotes and placeholder values are
dropped.
"""
before = self.paths()
self._list.clear()
paths = self._coerce(value)
for path in (paths[-1:] if self._single else paths):
self._append(path)
self._refresh_hint()
if self.paths() != before:
self.contents_changed.emit()
[docs]
def paths(self) -> List[str]:
"""Every path currently listed, in order -- always a list."""
return [self._list.item(row).data(Qt.UserRole)
for row in range(self._list.count())]
[docs]
def get_value(self) -> Any:
"""The setting's value: a ``list[str]``, or a ``str`` when single.
The shape is the SETTING's, not the widget's. A key declared ``str``
that came back as a one-element list rewrote the user's settings file
on open and reached ``pd.read_csv`` as a list.
"""
listed = self.paths()
if not self._single:
return listed
return listed[0] if listed else ""
_PLACEHOLDERS = {"list of paths", "none", "", "[]"}
@classmethod
def _coerce(cls, value: Any) -> List[str]:
"""Turn whatever the settings dict held into a list of paths.
ACCEPTS A BARE STRING as well as a list, because a setting that has
only ever held one path is written as one in older settings files.
:param value: the saved value.
:returns: the paths.
"""
if value is None:
return []
if isinstance(value, (str, bytes, os.PathLike)):
value = [value]
out: List[str] = []
for item in value:
if item is None:
continue
text = os.fspath(item) if isinstance(item, os.PathLike) else str(item)
text = text.strip().strip('"').strip("'")
if text.lower() in cls._PLACEHOLDERS:
continue
out.append(text)
return out
[docs]
def add_paths(self, paths: Iterable[Any]) -> int:
"""Append ``paths``, expanding folders. Returns how many were added.
When the setting names ONE file this REPLACES what is there. A second
choice is a correction, and a control that appended left the run
reading a file the user believed they had swapped out.
:param paths: paths to add, as strings or path-like objects; a bare
string counts as one path, and placeholders and None entries are
skipped. Each is made absolute, a folder contributes its matching
files one level down, and a single-file widget keeps only the last.
"""
incoming = self._coerce(paths)
if self._single:
if not incoming:
return 0
chosen = os.path.abspath(os.path.expanduser(incoming[-1]))
if chosen == (self.paths() or [None])[0]:
return 0
self._list.clear()
self._append(chosen)
self._refresh_hint()
self.value_changed.emit()
self.contents_changed.emit()
return 1
added = 0
for raw in incoming:
expanded = os.path.abspath(os.path.expanduser(raw))
if path_probe.isdir(expanded, wait=True):
for member in self._folder_members(expanded):
added += int(self._append(member))
else:
added += int(self._append(expanded))
if added:
self._refresh_hint()
self.value_changed.emit()
self.contents_changed.emit()
return added
def _folder_members(self, folder: str) -> List[str]:
"""Matching files one level inside ``folder``, sorted for stable order."""
extensions = _KIND_EXTENSIONS.get(self._kind, ())
try:
names = sorted(os.listdir(folder))
except OSError:
return []
members = []
for name in names:
full = os.path.join(folder, name)
if not os.path.isfile(full):
continue
if extensions and not name.lower().endswith(extensions):
continue
members.append(full)
return members
def _append(self, path: str) -> bool:
"""Add one path unless it is already listed. Returns True if added."""
resolved = os.path.abspath(os.path.expanduser(str(path)))
if resolved in set(self.paths()):
return False
item = QListWidgetItem(self._display_text(resolved))
item.setData(Qt.UserRole, resolved)
if path_probe.exists(resolved):
item.setToolTip(resolved)
else:
item.setToolTip(f"{resolved}\n\nThis path does not exist right now.")
item.setForeground(Qt.red)
self._list.addItem(item)
return True
@staticmethod
def _display_text(path: str) -> str:
"""Basename plus enough parent to tell four plate CSVs apart."""
parent = os.path.basename(os.path.dirname(path))
name = os.path.basename(path)
return f"{parent}/{name}" if parent else name
[docs]
def remove_selected(self) -> None:
"""Drop the selected paths.
Removed from the BOTTOM up, so each row index is still valid when it
is reached -- deleting top-down shifts everything below it.
"""
rows = sorted((self._list.row(item) for item in self._list.selectedItems()),
reverse=True)
for row in rows:
self._list.takeItem(row)
if rows:
self._refresh_hint()
self.value_changed.emit()
self.contents_changed.emit()
[docs]
def clear(self) -> None:
"""Drop every path, and say so only if there was anything to drop."""
if self._list.count():
self._list.clear()
self._refresh_hint()
self.value_changed.emit()
self.contents_changed.emit()
def _move_selected(self, offset: int) -> None:
"""Move the single selected row by ``offset``, keeping it selected."""
items = self._list.selectedItems()
if len(items) != 1:
return
row = self._list.row(items[0])
target = row + offset
if not 0 <= target < self._list.count():
return
item = self._list.takeItem(row)
self._list.insertItem(target, item)
self._list.setCurrentRow(target)
self.value_changed.emit()
self.contents_changed.emit()
def _empty_hint(self) -> str:
"""What to say when no paths have been added.
:returns: the hint text.
"""
if self._single:
return "Drop one file here, or use Choose file…"
return "Drop files or folders here, or use Add files…"
def _follow_path_probes(self) -> None:
"""Redraw the hint when a background path check finally answers.
`path_probe` reports an unknown path as PRESENT so the interface
never waits on a filesystem -- see its module docstring, and the
twenty-second `os.path.exists` that made it necessary. The cost of
that optimism is that a genuinely missing path is drawn as present
until the probe lands, so this is the half that corrects it.
Connected through a WEAK reference, and disconnected when the widget
is destroyed: the signal source is process-wide and outlives any one
widget. A closure over ``self`` connected for good kept every file
list ever built -- its whole Python wrapper tree -- alive for the
rest of the process, one more receiver on every probe answer each
time; a serial ``pytest tests/qt`` builds thousands. A destroyed C++
object behind a live Python wrapper is also what turns a redraw
into a hard crash, hence the ``RuntimeError`` guard.
"""
import weakref
from .. import path_probe as _probe
owner = weakref.ref(self)
def redraw(_path: str, _answer: bool) -> None:
"""Refresh the hint once a probe has an answer.
Both arguments are ignored: the hint is rebuilt from every path
it shows, so which one answered does not change the work.
:param _path: the path that was probed; unused.
:param _answer: what the probe found; unused.
"""
widget = owner()
if widget is None:
return
try:
widget._refresh_hint()
except RuntimeError:
pass
def let_go(*_args) -> None:
"""Drop the probe connection as the widget is destroyed.
:param _args: whatever ``destroyed`` sends; unused.
"""
try:
_probe.probes.answered.disconnect(redraw)
except (RuntimeError, TypeError):
pass
self._path_probe_redraw = redraw
_probe.probes.answered.connect(redraw)
self.destroyed.connect(let_go)
def _refresh_hint(self) -> None:
"""Show or hide the empty hint as the list changes.
Also mirrors the single-file field, because this is the one call
every mutation already makes -- set_value, a pick, a drop, a removal
and a path probe answering all reach here. Mirroring anywhere else
would be a second place to forget.
"""
if self._single_line is not None:
chosen = self.paths()
self._single_line.setText(chosen[0] if chosen else "")
self._single_line.setCursorPosition(0)
count = self._list.count()
missing = sum(
1 for row in range(count)
if not path_probe.exists(self._list.item(row).data(Qt.UserRole))
)
if not count:
self._hint.setText(self._empty_hint())
elif missing:
self._hint.setText(
f"{count} file{'s' if count != 1 else ''} selected — "
f"{missing} not found (shown in red)")
else:
self._hint.setText(
f"{count} file{'s' if count != 1 else ''} selected")
@staticmethod
def _urls(event) -> List[str]:
"""The local file paths a drag carries.
:param event: the Qt drag event.
:returns: the paths, empty when the drag carries none.
"""
mime = event.mimeData()
if not mime.hasUrls():
return []
return [url.toLocalFile() for url in mime.urls() if url.isLocalFile()]
[docs]
def dragEnterEvent(self, event: QDragEnterEvent) -> None: # noqa: N802
"""Accept a drag carrying file URLs.
:param event: the Qt drag event.
"""
if self._urls(event):
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dragMoveEvent(self, event: QDragMoveEvent) -> None: # noqa: N802
"""Keep accepting while file URLs stay over the list.
:param event: the Qt drag event.
"""
if self._urls(event):
event.acceptProposedAction()
else:
event.ignore()
[docs]
def dropEvent(self, event: QDropEvent) -> None: # noqa: N802
"""Add the dropped paths, ignoring a drop that carries none.
:param event: the Qt drop event.
"""
paths = self._urls(event)
if not paths:
event.ignore()
return
self.add_paths(paths)
event.acceptProposedAction()
[docs]
def pick_files(self) -> int:
"""Open the file dialog and take what is chosen.
Multi-select, except when the setting names one file -- a dialog that
lets you pick four when three of them will be discarded is a control
lying about what it does.
"""
if self._single:
path, _selected = QFileDialog.getOpenFileName(
self, self._title, self._start_directory(),
FILE_KIND_FILTERS[self._kind])
paths = [path] if path else []
else:
paths, _selected = QFileDialog.getOpenFileNames(
self, self._title, self._start_directory(),
FILE_KIND_FILTERS[self._kind])
if not paths:
return 0
self._last_directory = os.path.dirname(paths[0])
return self.add_paths(paths)
[docs]
def pick_folder(self) -> int:
"""Ask for a folder and add it.
:returns: how many paths were added; 0 when cancelled or duplicate.
"""
folder = QFileDialog.getExistingDirectory(
self, f"{self._title} — choose a folder", self._start_directory())
if not folder:
return 0
self._last_directory = folder
return self.add_paths([folder])
def _start_directory(self) -> str:
"""Reopen where the user last was, or beside the last file added."""
if self._last_directory and path_probe.isdir(self._last_directory,
wait=True):
return self._last_directory
values = self.paths()
if values:
parent = os.path.dirname(values[-1])
if path_probe.isdir(parent, wait=True):
return parent
return ""
__all__ = ["DATABASE_EXTENSIONS", "FilePathListWidget", "FILE_KIND_FILTERS",
"PairedFileTableWidget", "is_database_path", "side_for_header",
"suggest_file_pairs"]