Seitennummern in pdf einfügen

[mathi]

Da ich das Gefühl hatte, dass es it pool.map nix wird, habe ich mal versucht mit Process() weiterzukommen:

Code: Alles auswählen

# -*- coding: iso-8859-15 -*-

from pyPdf import PdfFileWriter, PdfFileReader
from tempfile import NamedTemporaryFile
from multiprocessing import Process, Queue 
import os 


def merge1(q1s,q1z): 
    while True:
        page=q1s.get()
        page.mergePage(q1z.get())    
        q1a.put(page)
        q1s.task_done()
        q1z.task_done()

if __name__ == '__main__':   
    output = PdfFileWriter()   
    input1 = PdfFileReader(file("test_lang.pdf", "rb")) 
    watermark = PdfFileReader(file("temp.pdf", "rb"))  
    pages=input1.getNumPages()
    
    q1s=Queue() #Seiten=input1
    q1z=Queue() #Seitenzahlen = watermark
    q1a=Queue() #Output

    for i in xrange(pages):
        q1s.put(input1.getPage(i))
        q1z.put(watermark.getPage(i))
              
    p1 = Process(target=merge1, args=(q1s,q1z)) 
    p1.start() 
    p1.join()

    while True:
        page=q1a.get()
        output.addPage(page)
                 
    temp_file = NamedTemporaryFile(suffix = '.pdf', prefix = 'temp',delete=False) 
    output.write(temp_file) 
    os.startfile(temp_file.name) 
    temp_file.close() 

Ich haber erstmal nur einen Prozess, um zu sehen ob es funktioniert...die Fehlermeldung lautet:

Code: Alles auswählen

Process Process-1:
Traceback (most recent call last):
  File "C:\Python26\lib\multiprocessing\process.py", line 232, in _bootstrap
    self.run()
  File "C:\Python26\lib\multiprocessing\process.py", line 88, in run
    self._target(*self._args, **self._kwargs)
  File "D:\Python\Code\wxPython\workspace\pdfPage\src\pdfPageCMD5a.py", line 12, in merge1
    page.mergePage(q1z.get())
  File "C:\Python26\lib\site-packages\pyPdf\pdf.py", line 1095, in mergePage
    self._mergePage(page2)
  File "C:\Python26\lib\site-packages\pyPdf\pdf.py", line 1122, in _mergePage
    new, newrename = PageObject._mergeResources(originalResources, page2Resources, res)
  File "C:\Python26\lib\site-packages\pyPdf\pdf.py", line 1030, in _mergeResources
    page2Res = res2.get(resource, DictionaryObject()).getObject()
  File "C:\Python26\lib\site-packages\pyPdf\generic.py", line 165, in getObject
    return self.pdf.getObject(self).getObject()
  File "C:\Python26\lib\site-packages\pyPdf\pdf.py", line 615, in getObject
    self.stream.seek(start, 0)
ValueError: I/O operation on closed file

da ich aber sämtliche Seiten in je eine Queue gepackt habe, kann

Offensichtlich werden die Dateien zwischenzeitlich geschlossen.

doch nicht stimmen...

[HWK]

page enthält wohl nicht den Seiteninhalt der pdf-Datei, sondern nur eine Referenz darauf. Der Inhalt muss also erst wieder aus der Datei ausgelesen werden, die aber offensichtlich schon wieder geschlossen wurde. Auch zum Schreiben der Ausgabe muss wohl die Eingabedatei geöffnet sein. Versuch doch mal statt

Code: Alles auswählen

input1 = PdfFileReader(file("test_lang.pdf", "rb"))

Code: Alles auswählen

with open('test_lang.pdf', 'rb') as infile:
    input1 = PdfFileReader(infile)

um den GC am Schließen der datei zu hindern. Damit kannst Du ja mal Deine verscheidenen Varianten testen.
MfG
HWK

[mathi]

@HWK.
vielen Dank für Deine Geduld, ich bin froh, dass ich Hilfe bekomme

zusammengefasst kann man sagen, es ändert sich nichts
Hier mal die vielversprechendste Version:http://paste.pocoo.org/show/XcyNU3ej5Bk8zEOVwqPb/

es werden 4 python.exe im Tastmanager angezeigt, es wird aber nichts gerechnet...

in der Version mit Process() http://paste.pocoo.org/show/ruDIdmowrkJixVQqy1B0/
bleibt die Fehlermeldung.

Kann es sein, dass es an pypdf liegt??

[HWK]

Das wird zwar das Problem nicht lösen, aber neue vermeiden:
Du solltest nur eine Queue verwenden, in der Du Tuple aus der Seitennummer, der Originalseite und der "Partnerseite" speicherst, damit Du sicher sein kannst die richtigen Seiten zu verknüpfen und deren Seitennummer kennst. merge sollte ein Tuple aus Seitennummer und der fertigen Seiten zurückliefern. Zum Schluss solltest Du die Seiten in der Reihenfolge ihrer Seitennummer ausgeben.
Um den Fehler einzugrenzen, solltest Du ein paar informative Prints einbauen, z.B. in merge pagenum und page ausgeben lassen etc. Dann kann man weiter sehen.
Ich möchte wohl verständlicherweise meinen Rechner nicht mit Modulen "zuzumüllen", die ich ansonsten nicht brauche, um Deine Scripts zu testen.
MfG
HWK

[mathi]

so, ich habs mal so gemacht, wie vorgeschlagen (außer der Ergebnistupelausgabe)
hier mal dasergebnis: http://paste.pocoo.org/show/jE9BOITC2OJWdtqX2cRk/

als singleprocess funktioniert es (natürlich),

aber es hängt beim start der pool.map, denn hier die Ausgabe:

Code: Alles auswählen

test1
test1.1

es wird die Zeile

Code: Alles auswählen

maplist= pool.map(merge,xrange(pages))

gar nicht erst ausgeführt, d.h. merge nicht ein einziges mal aufgerufen...

übrigens das auskommentierte

Code: Alles auswählen

print result.get(timeout=10) 

bringt einen timeout

[HWK]

Code: Alles auswählen

pool.apply_async

ruft merge auch nicht auf?
MfG
HWK

[mathi]

nein, bzw. wird das einfach übersprungen - ich bekomme zwar keinen Fehler, aber die print-Anweisungen in merge werden nicht ausgeführt...

das print des ergebnisses von pool.apply_async bringt dann den timeout (auch bei einem Wert von timeout=100)

[HWK]

Schreib doch die "Print-Nachrichten" in merge einmal in eine Log-Datei.
MfG
HWK

[mathi]

wir kommen der Sache näher...

Code: Alles auswählen

def merge(pagenum): 
    print "pagenum = %s" % pagenum
    log=file('logfile%s.txt' % pagenum,'w')     
    log.write("pagenum = %s" % pagenum)
    #log.close()  #----- erzeugt txt. Dateien mit "pagenum= 7" usw.
    pagelist=q.get()
    log.write(pagelist)
    log.close() #---- hier bleibt es hängen, d.h. q.get() macht Probleme
    page=pagelist[1] 
    watermark=pagelist[2]
    page.mergePage(watermark) 
    return page
    q.task_done()

merge wird also doch aufgerufen, es entstehen 4 .txt Dateien (logfile0.txt, logfile7.txt, logfile14.txt, logfile21.txt) mit log.close() vor q.get() steht in den Dateien jeweils die Seitennummer, bei log.close() nach q.get() bleiben die Dateien leer...

(wobei es mich wundert, dass print vor q.get() nicht ausgeführt wird)

[HWK]

Noch eine Idee:
Übergib doch q mal merge als Argument. pagenum brauchst Du nicht als Argument. Die richtige Seitennummer steht ja in der Queue. q braucht dann auch nicht mehr global zu sein.
MfG
HWK

[mathi]

geht nicht so einfach, weil pool.map() zwingend das iterable an merge übergibt,

also habe ich versucht functools.partial wieder zu verwenden
http://paste.pocoo.org/show/xDdF2XnLoKMmibYPBUwo/
, aber das wirft folgenden Fehler aus:

Code: Alles auswählen

Traceback (most recent call last):
  File "C:\Program Files\eclipse\plugins\org.python.pydev_1.5.2.1260362205\PySrc\pydev_sitecustomize\sitecustomize.py", line 142, in <module>
    sys.path.extend(paths_removed)
AttributeError: 'NoneType' object has no attribute 'path'
Exception in thread Thread-1:
Traceback (most recent call last):
  File "C:\Python26\lib\threading.py", line 525, in __bootstrap_inner
    self.run()
  File "C:\Python26\lib\threading.py", line 477, in run
    self.__target(*self.__args, **self.__kwargs)
  File "C:\Python26\lib\multiprocessing\pool.py", line 225, in _handle_tasks
    put(task)
  File "C:\Python26\lib\multiprocessing\queues.py", line 279, in __getstate__
    return Queue.__getstate__(self) + (self._cond, self._unfinished_tasks)
  File "C:\Python26\lib\multiprocessing\queues.py", line 51, in __getstate__
    assert_spawning(self)
  File "C:\Python26\lib\multiprocessing\forking.py", line 25, in assert_spawning
    ' through inheritance' % type(self).__name__
RuntimeError: JoinableQueue objects should only be shared between processes through inheritance

Vererbung?? Ich hab keine Ahnung wie das hier gehen soll...
oder habe ich eine andere Möglichkeit q zu übergeben??

[HWK]

Als iterable

Code: Alles auswählen

[q] * anzahl_der_seiten

MfG
HWK

[mathi]

mit

Code: Alles auswählen

maplist= pool.map(merge,([q]*pages)) 

folgt der selbe Fehler:

Code: Alles auswählen

  File "C:\Python26\lib\multiprocessing\forking.py", line 25, in assert_spawning
    ' through inheritance' % type(self).__name__
RuntimeError: JoinableQueue objects should only be shared between processes through inheritance

[mathi]

Hallo,
ich möchte nochmal um Hilfe bitten, da ich das Problem nicht alleine lösen kann......

[HWK]

Vielleicht hilft das? D.h. versuch mal

Code: Alles auswählen

    manager = multiprocessing.Manager()
    q = manager.Queue()

MfG
HWK

[HWK]

Falls das auch nicht klappt, hier mein nächster Vorschlag:
Verzichte auf die Queue, übergib an merge die Seitennummer und diesen pdf_reader statt input1.

Code: Alles auswählen

...
class MyManager(BaseManager):
    pass

MyManager.register('Reader', pyPdf.PdfFileReader)
...
    manager = MyManager()
    manager.start()
    pdf_reader = manager.Reader(open('test_lang.pdf', 'rb'))
...

Lies dann in merge die Seiten ein, berechne das Wasserzeichen bzw. benutze zum Testen einen zweiten vergleichbaren
Reader und verbinde beide Seiten. Das Ergebnis ist der Rückgabewert.
MfG
HWK

[mathi]

hallo und Danke für die Antworten,

von den beiden Varianten funktioniert erstere wohl am "Besten"
http://paste.pocoo.org/show/XeOI1YncBh9HVY7LjVWN/

wobei das nur mit Lock() einigermaßen funktioniert....
mit Lock() - pool.map() scheint eine Weile zu laufen :

Code: Alles auswählen

Traceback (most recent call last):
  File "D:\Python\Code\wxPython\workspace\pdfPage\src\pdfPageCMD.py", line 48, in <module>
    main()
  File "D:\Python\Code\wxPython\workspace\pdfPage\src\pdfPageCMD.py", line 27, in main
    maplist= pool.map(merge,([q]*pages,lock))          # so geht's nicht
  File "C:\Python26\lib\multiprocessing\pool.py", line 148, in map
    return self.map_async(func, iterable, chunksize).get()
  File "C:\Python26\lib\multiprocessing\pool.py", line 422, in get
    raise self._value
AttributeError: 'list' object has no attribute 'get'

mich wundert, dass dabei nur 1 Prozess mit 2 threads gestartet wird, da je lock ein threadingobjekt erstellt wird, aber keine 4 Prozesse

ohne Lock() das altbekannte:

Code: Alles auswählen

ValueError: I/O operation on closed file

[HWK]

Was Du da tust, weißt Du nicht wirklich, oder? Was soll denn das Lock? An merge übergibst Du ein Tuple aus einer Liste von q's und dem Lock. Die Liste hat natürlich keine get-Methode. Ich denke, bevor Du Dich mit hochkomplexen Sachverhalten wie der Multiprozess-Programmierung beschäftigst, solltest Du Dir erst noch einmal die Grundlagen aneignen. Zumal nicht einmal klar ist, ob Multiprocessing die Performance Deiner Anwendung verbessern wird.
MfG
HWK

[mathi]

@HWK
hast recht

Code: Alles auswählen

    t1=time()   
    pages=input1.getNumPages()

    pool = Pool(processes=4)
    maplist= pool.map(merge,xrange(pages))         
    pool.close() 
    pool.join()
    
    print "länge=",len(maplist)
    t2=time()
    print "time=",(t2-t1)

dauert 28 sec.

Code: Alles auswählen

    t1=time() 
    output = PdfFileWriter()  
            
    with open('test_lang.pdf', 'rb') as infile: 
        input1 = PdfFileReader(infile) 
        watermark = PdfFileReader(file("temp.pdf", "rb"))  
        pages=input1.getNumPages()

        pagelist=[]    
        for pagenum in range(pages):                  
            pagelist.append(merge(pagenum))
        t2=time()
        print "time=",(t2-t1)

dauert 22 sec.

Hätte ich nicht gedacht...

ich habe jetzt herausgefunden, dass gar nicht map() schuld ist, sondern

Code: Alles auswählen

    for page in maplist:
        output.addPage(page)

ich kann nicht einmal

Code: Alles auswählen

output.addPage(maplist[0])

ausführen.

Somit konnte ich aber zumindest die Zeitmessung vornehmen.

Das bedeutet doch, dass mein Prog. eher I/O lastig ist, aber die HDD schreibt nix, daher bleibt alles im RAM.
Gibt es überhaupt eine Möglichkeit mein Programm zu beschleunigen??

[mathi]

Da ich der Meinung bin, dass das Problem darin bestand die Ergebnisse von

Code: Alles auswählen

mergePage

fehlerfrei aus der merge()-Funktionen herauszubekommen, habe ich eine Möglichkeit gefunden

Ich teile die Arbeit in (bei 4 Kernen) 4 teile und erstelle in 4 merge-Funktionen je 1/4 der Seiten und erhalte insgesamt 4 .pdf-Dateien, die dann noch zu einer zusammengeführt werden sollen.

http://paste.pocoo.org/show/wBopIiVEKqi34m6KDYvy/

Der Code ist noch ungefeilt, aber es funktioniert soweit, dass mit x Kernen auch die einzelne Arbeit verrichtet wird und die Dateien in der richtigen Reihenfolge zusammengefügt werden

zum Test: ein Kern braucht hier 33 sec. , 4 Kerne schaffen das in 14 sec.
bei einer .pdf mit 110 Seiten, interessant wird es dann mit 500-1200 Seiten, die bei mir öffter vorkommen


Ich würde mich über Eure Tipps freuen, den Code zu verbessern