Obtenir de l'avancement des travaux de shutil de copie de fichier thread

J'ai une application à partir de laquelle un fichier est copié à partir de src à dst:

import shutil
from threading import Thread

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

Je souhaite avoir l'application de requête de la progression de la copie toutes les 5 secondes sans bloquer l'application elle-même. Est-ce possible?

Mon intention est de définir ce progrès pour un QtGui.QLabel pour donner à l'utilisateur un retour sur la copie de fichier.

Cela peut-il être réalisé lors de la copie à l'aide d'une tige filetée shutil de copie de fichier?

InformationsquelleAutor fredrik | 2015-04-30
  1. 18

    shutil.copy() n'offre pas toutes les options pour suivre les progrès, pas de. Tout au plus on pourrait contrôler la taille du fichier de destination (à l'aide de os.* fonctions sur le nom de fichier cible).

    L'alternative consisterait à mettre en place votre propre fonction de copie. L'application est vraiment très simple; shutil.copier() est fondamentalement un shutil.copyfile() plus shutil.copymode() appel; shutil.copyfile() à son tour, délègue le travail réel à shutil.copyfileobj() (des liens vers le code source Python).

    La mise en œuvre de votre propre shutil.copyfileobj() d'inclure des progrès devraient être trivial; injecter de soutien pour une fonction de rappel pour le rapport d'informer votre programme chaque fois qu'un autre bloc a copié:

    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)

    et de comparer les copied taille avec la taille du fichier.

    • Merci pour cette! Je l'ai fait avec les autres fonctions nécessaires à partir de shutil et une barre de progression de fonction pour créer une solution complète que j'ai posté comme une autre réponse.
    InformationsquelleAutor Martijn Pieters
  2. 4

    J'ai combiné Martijn Pieters répondre des progrès de code à barres de cette réponse avec des modifications de travailler dans PyCharm de cette réponse qui me donne ce qui suit. La fonction copy_with_progress était mon objectif.

    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
    InformationsquelleAutor flutefreak7
  3. 0

    Non, il ne peut pas être fait de cette façon, parce que shutil.copy n'ont pas les moyens de fournir progrès.

    Mais vous pouvez écrire votre propre fonction de copie (ou même un fork du code de shutil--avis que c'est l'un des modules qui comprend un lien vers la source au sommet, ce qui signifie qu'il est destiné à être aussi utile pour l'exemple de code pour pour l'utiliser en tant que-est). Votre fonction peut, par exemple, prendre un progrès fonction de rappel comme un argument supplémentaire et l'appelle après chaque tampon (ou chaque N tampons, ou chaque N octets, ou toutes les N secondes). Quelque chose comme:

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

    Maintenant, que le rappel est toujours en cours d'être appelé dans le thread d'arrière-plan, pas le thread principal. Avec la plupart des GUI cadres, ce qui signifie qu'il ne peut pas toucher directement l'interface graphique widgets. Mais la plupart des GUI cadres ont une façon de poster un message sur le thread principal de la boucle d'événements à partir d'un thread d'arrière-plan, il suffit donc de faire le rappel de le faire. Avec Qt-vous faire cela avec les signaux et les slots, exactement de la même façon que dans le thread principal; il y a beaucoup de tutoriels là-bas si vous ne savez pas comment.

    Sinon, vous pouvez le faire de la manière que vous avez suggéré: avoir le thread principal signal le thread d'arrière-plan (par exemple, en postant sur un queue.Queue ou le déclenchement d'une Event ou Condition) et avoir votre copy fonction de vérifier que le signal de tous les temps à travers la boucle et d'y répondre. Mais qui semble à la fois plus complexe et moins réactif.

    Une chose de plus: Qt a son propre filetage de la bibliothèque, et vous pouvez l'utiliser au lieu de Python natif de l'un, parce que vous pouvez joindre une fente directement à QThread objet et assurez-vous que votre rappel. Je ne suis pas sûr, mais Qt pourrait même avoir sa propre copie de fichier-avec-les progrès des méthodes de quelque part; ils essaient de s'envelopper tout qui pourrait être à tous les différents entre les plates-formes et vaguement liés aux Interfaces graphiques.

    InformationsquelleAutor abarnert
  4. 0

    En plus de Martijn Pieters excellente réponse, si (comme moi, Je suis un idiot) vous avez besoin de savoir comment passer la réelle rappel dans le copyfileobj() fonction, vous pouvez le faire comme ceci:

    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)
...
    InformationsquelleAutor Dhiraj Gupta
  5. 0

    Cela peut être très orthodoxe, mais il fonctionne:

    """
Copying a file and checking its progress while it's copying.
"""
import os
import shutil
import threading
import time
des = r'<PATH/TO/SPURCE/FILE>'
src = r'<PATH/TO/DESTINATION/FILE>'
def checker(source_path, destination_path):
"""
Compare 2 files till they're the same and print the progress.
:type source_path: str
:param source_path: path to the source file
:type destination_path: str
:param destination_path: path to the destination file
"""
# Making sure the destination path exists
while not os.path.exists(destination_path):
print "not exists"
time.sleep(.01)
# Keep checking the file size till it's the same as source file
while os.path.getsize(source_path) != os.path.getsize(destination_path):
print "percentage", int((float(os.path.getsize(destination_path))/float(os.path.getsize(source_path))) * 100)
time.sleep(.01)
print "percentage", 100
def copying_file(source_path, destination_path):
"""
Copying a file
:type source_path: str
:param source_path: path to the file that needs to be copied
:type destination_path: str
:param destination_path: path to where the file is going to be copied
:rtype: bool
:return: True if the file copied successfully, False otherwise
"""
print "Copying...."
shutil.copyfile(source_path, destination_path)
if os.path.exists(destination_path):
print "Done...."
return True
print "Filed..."
return False
t = threading.Thread(name='copying', target=copying_file, args=(src, des))
# Start the copying on a separate thread
t.start()
# Checking the status of destination file on a separate thread
b = threading.Thread(name='checking', target=checker, args=(src, des))
b.start()
    InformationsquelleAutor ali jafargholi