"""A background transfer queue with progress reporting. Jobs (copy / download / dump) run one at a time on a worker thread and report progress back to the GUI via Qt signals. """ import logging import threading from PySide6.QtCore import QObject, QThread, Signal from .store import TransferCancelled log = logging.getLogger("queue") class TransferJob: _ids = 0 def __init__(self, kind, title, source_label, dest): TransferJob._ids += 1 self.id = TransferJob._ids self.kind = kind self.title = title self.source_label = source_label self.dest = str(dest) self.item = None self.source = None self.dest_dir = None self.status = "queued" # queued/running/done/skipped/cancelled/cancelling/error self.stage = "" self.detail = "" self.progress = 0.0 self.error = "" def to_dict(self): return { "id": self.id, "kind": self.kind, "title": self.title, "source": self.source_label, "dest": self.dest, "status": self.status, "stage": self.stage, "detail": self.detail, "progress": self.progress, "error": self.error, } class JobWorker(QObject): progress = Signal(int, str, float, str) # job_id, stage, fraction, detail finished = Signal(int, object) # job_id, result skipped = Signal(int, str) # job_id, path (already present) cancelled = Signal(int) # job_id failed = Signal(int, str) # job_id, error def __init__(self, source, dest_dir, job): super().__init__() self.source = source self.dest_dir = dest_dir self.job = job self._cancel = threading.Event() def cancel(self): self._cancel.set() def _progress(self, stage, fraction, detail=""): self.progress.emit(self.job.id, stage, fraction, detail) def run(self): try: result = self.source.download(self.job.item, self.dest_dir, self._progress, self._cancel) self.finished.emit(self.job.id, result) except TransferCancelled: self.cancelled.emit(self.job.id) except FileExistsError as exc: self.skipped.emit(self.job.id, str(exc)) except Exception as exc: # noqa: BLE001 self.failed.emit(self.job.id, str(exc)) class TransferQueue(QObject): job_added = Signal(dict) job_updated = Signal(dict) job_finished = Signal(dict) job_removed = Signal(int) def __init__(self, parent=None): super().__init__(parent) self.jobs = [] self._pending = [] self._current = None self._thread = None self._worker = None # Keep strong references to every worker/thread for their full lifetime. # Dropping a Python ref to a QObject that was moved to another thread # deletes it directly and crashes (bus error). self._threads = [] self._workers = [] def enqueue(self, source, dest_dir, item, title, kind="download"): job = TransferJob(kind, title, getattr(source, "label", "source"), dest_dir) job.item = item job.source = source job.dest_dir = dest_dir self.jobs.append(job) self._pending.append(job) log.info("queued %s (%s) -> %s", title, kind, dest_dir) self.job_added.emit(job.to_dict()) self._start_next() return job def cancel(self, job_id): job = self._find(job_id) if job is None: return if job.status == "queued": if job in self._pending: self._pending.remove(job) job.status = "cancelled" self.job_updated.emit(job.to_dict()) self.job_finished.emit(job.to_dict()) return if job.status == "running" and self._worker is not None: self._worker.cancel() job.status = "cancelling" self.job_updated.emit(job.to_dict()) def remove(self, job_id): """Drop a job from the history (active jobs must be cancelled first).""" job = self._find(job_id) if job is None: return if job.status in ("running", "cancelling"): return if job in self._pending: self._pending.remove(job) self.jobs.remove(job) self.job_removed.emit(job_id) def clear_finished(self): """Remove all finished jobs from the history.""" finished = [j.id for j in self.jobs if j.status in ("done", "skipped", "cancelled", "error")] for jid in finished: self.remove(jid) def shutdown(self): if self._worker is not None: self._worker.cancel() for t in self._threads: try: if t.isRunning(): t.quit() t.wait() except RuntimeError: pass def _find(self, job_id): for j in self.jobs: if j.id == job_id: return j return None def _start_next(self): if self._current is not None or not self._pending: return job = self._pending.pop(0) self._current = job job.status = "running" log.info("starting job #%d: %s (%s)", job.id, job.title, job.kind) self.job_updated.emit(job.to_dict()) self._thread = QThread(self) self._worker = JobWorker(job.source, job.dest_dir, job) self._worker.moveToThread(self._thread) self._thread.started.connect(self._worker.run) self._worker.progress.connect(self._on_progress) self._worker.finished.connect(self._on_finished) self._worker.skipped.connect(self._on_skipped) self._worker.cancelled.connect(self._on_cancelled) self._worker.failed.connect(self._on_failed) for sig in (self._worker.finished, self._worker.skipped, self._worker.cancelled, self._worker.failed): sig.connect(self._thread.quit) self._thread.finished.connect(self._on_thread_finished) self._threads.append(self._thread) self._workers.append(self._worker) self._thread.start() def _on_progress(self, job_id, stage, fraction, detail): job = self._find(job_id) if job is None: return job.stage = stage job.progress = fraction job.detail = detail self.job_updated.emit(job.to_dict()) def _on_finished(self, job_id, result): self._finish_job(job_id, "done", detail=str(result) if result is not None else "") def _on_skipped(self, job_id, path): self._finish_job(job_id, "skipped", error=path) def _on_cancelled(self, job_id): self._finish_job(job_id, "cancelled") def _on_failed(self, job_id, error): self._finish_job(job_id, "error", error=error) def _finish_job(self, job_id, status, detail="", error=""): job = self._find(job_id) if job is not None: job.status = status if detail: job.detail = detail if error: job.error = error if status == "done": job.progress = 1.0 log.info("job #%d %s: %s%s", job.id, status, job.title, f" — {error}" if error else "") self.job_finished.emit(job.to_dict()) def _on_thread_finished(self): # Queued to the main thread once the worker thread has fully stopped, # so it is now safe to drop the "current" pointers and advance. self._current = None self._thread = None self._worker = None self._start_next()