Fortschritt zurück aus shutil Datei kopieren Thread

Question

ich eine Anwendung haben, aus dem eine Datei src-dst kopiert wird:

import shutil 
from threading import Thread 

t = Thread(target=shutil.copy, args=[ src, dst ]).start() 

Ich wünsche der Anwendung Abfrage haben, um den Fortschritt der Kopie alle 5 Sekunden ohne die Anwendung selbst zu sperren. Ist das möglich?

Ich möchte diesen Fortschritt auf QtGui.QLabel setzen, um dem Benutzer Feedback zur Dateikopie zu geben.

Kann dies beim Kopieren mit einer Shutfile-Kopie mit Thread erreicht werden?

Answer 1

shutil.copy() bietet keine Optionen zum Verfolgen des Fortschritts, nein. Sie können höchstens die Größe der Zieldatei überwachen (mit os.* Funktionen auf den Zieldateinamen).

Die Alternative wäre, Ihre eigene Kopierfunktion zu implementieren. Die Implementierung ist wirklich ziemlich einfach; shutil.copy() ist im Grunde eine shutil.copyfile() plus shutil.copymode() Anruf; shutil.copyfile() wiederum delegiert die eigentliche Arbeit an shutil.copyfileobj() (Links zum Python-Quellcode).

Die Implementierung Ihres eigenen shutil.copyfileobj(), um den Fortschritt einzuschließen, sollte trivial sein; injizieren Unterstützung für eine Callback-Funktion zu berichten Ihr Programm jedes Mal ein anderer Block informieren, kopiert:

def copyfileobj(fsrc, fdst, callback, length=16*1024): 
    copied = 0 
    while True: 
     buf = fsrc.read(length) 
     if not buf: 
      break 
     fdst.write(buf) 
     copied += len(buf) 
     callback(copied) 

und vergleichen Sie die copied Größe der Dateigröße.

Answer 2

Nein, es kann nicht auf diese Weise gemacht werden, weil shutil.copy keine Möglichkeit hat, den Fortschritt zu liefern.

Aber Sie können Ihre eigene Kopierfunktion (oder sogar Gabel den Code aus shutil --notice schreiben, dass es eines der Module ist, die einen Link zu the source an der Spitze enthält, es soll Sinn für Beispielcode als nützlich sein als für nur verwenden, wie es ist). Ihre Funktion kann z. B. eine Fortschrittsrückruffunktion als zusätzliches Argument annehmen und sie nach jedem Puffer (oder jedem N Puffer oder jedem N Byte oder jedem N Sekunden) aufrufen. Etwas wie:

def copy(src, dst, progress): 
    # ... 
    for something: 
     progress(bytes_so_far, bytes_total) 
     # ... 
    progress(bytes_total, bytes_total) 

Jetzt wird dieser Callback immer noch im Hintergrund Thread, nicht der Haupt-Thread aufgerufen werden. Bei den meisten GUI-Frameworks bedeutet dies, dass GUI-Widgets nicht direkt berührt werden können. Aber die meisten GUI-Frameworks haben die Möglichkeit, eine Nachricht von einem Hintergrundthread an die Ereignisschleife des Hauptthreads zu senden, also machen Sie das Callback einfach. Mit Qt machen Sie das mit Signalen und Slots, genauso wie Sie es im Hauptthread tun; Es gibt viele großartige Tutorials, wenn Sie nicht wissen, wie.

Alternativ Sie es die Art und Weise tun, könnten Sie vorgeschlagen: haben die Haupt-Thread die Hintergrund-Thread-Signal (zB durch eine queue.Queue veröffentlichen oder eine Event oder Condition Triggerung) für dieses Signal und haben Ihre copy Funktionsprüfung jedes Mal durch die Schleife und reagieren. Aber das scheint sowohl komplizierter als auch weniger ansprechend zu sein.

Noch eine Sache: Qt hat seine eigene Threading-Bibliothek, und Sie können sie anstelle von Pythons nativen verwenden, weil Sie einen Slot direkt an QThread-Objekt anfügen können und das zu Ihrem Rückruf machen. Ich bin mir nicht sicher, aber Qt könnte sogar irgendwo eine eigene Datei-Copy-with-Progress-Methode haben; sie versuchen, alles zu wickeln, das zwischen Plattformen und vage verwandt mit GUIs überhaupt unterschiedlich sein könnte.

Answer 3

Neben Martijn Pieters ausgezeichnete Antwort, wenn (wie ich, I'm an idiot) Sie brauchen, um herauszufinden, wie der tatsächlichen Rückruf in die copyfileobj() Funktion zu übergeben, können Sie es wie folgt tun:

def myscopefunction(): 
    ### Inside wherever you want to call the copyfileobj() function, you can 
    ### make a nested function like so: 
    def progress(bytescopied): 
     updateui(bytescopied) #update your progress bar or whatever 

    #and then call it like this 
    copyfileobj(source,destination,progress) 
    ... 

Answer 4

I kombiniert mit einem Fortschrittsbalkencode von this answer mit Änderungen in PyCharm von this answer zu arbeiten, die mir folgendes gibt. Die Funktion copy_with_progress war mein Ziel.

import os 
import shutil 


def progress_percentage(perc, width=None): 
    # This will only work for python 3.3+ due to use of 
    # os.get_terminal_size the print function etc. 

    FULL_BLOCK = '█' 
    # this is a gradient of incompleteness 
    INCOMPLETE_BLOCK_GRAD = ['░', '▒', '▓'] 

    assert(isinstance(perc, float)) 
    assert(0. <= perc <= 100.) 
    # if width unset use full terminal 
    if width is None: 
     width = os.get_terminal_size().columns 
    # progress bar is block_widget separator perc_widget : ####### 30% 
    max_perc_widget = '[100.00%]' # 100% is max 
    separator = ' ' 
    blocks_widget_width = width - len(separator) - len(max_perc_widget) 
    assert(blocks_widget_width >= 10) # not very meaningful if not 
    perc_per_block = 100.0/blocks_widget_width 
    # epsilon is the sensitivity of rendering a gradient block 
    epsilon = 1e-6 
    # number of blocks that should be represented as complete 
    full_blocks = int((perc + epsilon)/perc_per_block) 
    # the rest are "incomplete" 
    empty_blocks = blocks_widget_width - full_blocks 

    # build blocks widget 
    blocks_widget = ([FULL_BLOCK] * full_blocks) 
    blocks_widget.extend([INCOMPLETE_BLOCK_GRAD[0]] * empty_blocks) 
    # marginal case - remainder due to how granular our blocks are 
    remainder = perc - full_blocks*perc_per_block 
    # epsilon needed for rounding errors (check would be != 0.) 
    # based on reminder modify first empty block shading 
    # depending on remainder 
    if remainder > epsilon: 
     grad_index = int((len(INCOMPLETE_BLOCK_GRAD) * remainder)/perc_per_block) 
     blocks_widget[full_blocks] = INCOMPLETE_BLOCK_GRAD[grad_index] 

    # build perc widget 
    str_perc = '%.2f' % perc 
    # -1 because the percentage sign is not included 
    perc_widget = '[%s%%]' % str_perc.ljust(len(max_perc_widget) - 3) 

    # form progressbar 
    progress_bar = '%s%s%s' % (''.join(blocks_widget), separator, perc_widget) 
    # return progressbar as string 
    return ''.join(progress_bar) 


def copy_progress(copied, total): 
    print('\r' + progress_percentage(100*copied/total, width=30), end='') 


def copyfile(src, dst, *, follow_symlinks=True): 
    """Copy data from src to dst. 

    If follow_symlinks is not set and src is a symbolic link, a new 
    symlink will be created instead of copying the file it points to. 

    """ 
    if shutil._samefile(src, dst): 
     raise shutil.SameFileError("{!r} and {!r} are the same file".format(src, dst)) 

    for fn in [src, dst]: 
     try: 
      st = os.stat(fn) 
     except OSError: 
      # File most likely does not exist 
      pass 
     else: 
      # XXX What about other special files? (sockets, devices...) 
      if shutil.stat.S_ISFIFO(st.st_mode): 
       raise shutil.SpecialFileError("`%s` is a named pipe" % fn) 

    if not follow_symlinks and os.path.islink(src): 
     os.symlink(os.readlink(src), dst) 
    else: 
     size = os.stat(src).st_size 
     with open(src, 'rb') as fsrc: 
      with open(dst, 'wb') as fdst: 
       copyfileobj(fsrc, fdst, callback=copy_progress, total=size) 
    return dst 


def copyfileobj(fsrc, fdst, callback, total, length=16*1024): 
    copied = 0 
    while True: 
     buf = fsrc.read(length) 
     if not buf: 
      break 
     fdst.write(buf) 
     copied += len(buf) 
     callback(copied, total=total) 


def copy_with_progress(src, dst, *, follow_symlinks=True): 
    if os.path.isdir(dst): 
     dst = os.path.join(dst, os.path.basename(src)) 
    copyfile(src, dst, follow_symlinks=follow_symlinks) 
    shutil.copymode(src, dst) 
    return dst