Initial commit
This commit is contained in:
+227
@@ -0,0 +1,227 @@
|
||||
"""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()
|
||||
|
||||
Reference in New Issue
Block a user