228 lines
7.6 KiB
Python
228 lines
7.6 KiB
Python
"""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()
|
|
|