Could not create profile. Error:
%s'%str(err)).exec_()
@@ -146,7 +136,7 @@ class %(classname)s(%(base_class)s):
else:
src = qstring_to_unicode(self.source_code.toPlainText())
try:
- title = self.create_class(src).title
+ title = create_class(src).title
except Exception, err:
error_dialog(self, 'Invalid input',
'
Could not create profile. Error:
%s'%str(err)).exec_()
diff --git a/src/libprs500/gui2/jobs.py b/src/libprs500/gui2/jobs.py
index 8b0a0215d7..4a3315a465 100644
--- a/src/libprs500/gui2/jobs.py
+++ b/src/libprs500/gui2/jobs.py
@@ -12,13 +12,14 @@
## You should have received a copy of the GNU General Public License along
## with this program; if not, write to the Free Software Foundation, Inc.,
## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
-import traceback, logging, cStringIO
+import traceback, logging
from PyQt4.QtCore import QAbstractTableModel, QMutex, QObject, SIGNAL, Qt, \
QVariant, QThread, QModelIndex, QSettings
from PyQt4.QtGui import QIcon
from libprs500.gui2 import NONE
+from libprs500.parallel import Server
class JobException(Exception):
pass
@@ -44,11 +45,7 @@ class Job(QThread):
self.logger = logging.getLogger('Job #'+str(id))
self.logger.setLevel(logging.DEBUG)
self.is_locked = False
- self.log_dest = cStringIO.StringIO()
- handler = logging.StreamHandler(self.log_dest)
- handler.setLevel(logging.DEBUG)
- handler.setFormatter(logging.Formatter('[%(levelname)s] %(filename)s:%(lineno)s: %(message)s'))
- self.logger.addHandler(handler)
+ self.log = None
def lock(self):
@@ -85,23 +82,23 @@ class DeviceJob(Job):
self.emit(SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
self.id, self.description, self.result, exception, last_traceback)
-
+MPServer = None
class ConversionJob(Job):
- ''' Jobs that invlove conversion of content. Synchronous. '''
+ ''' Jobs that involve conversion of content. Synchronous. '''
def run(self):
self.lock()
last_traceback, exception = None, None
try:
try:
- self.kwargs['logger'] = self.logger
- self.result = self.func(*self.args, **self.kwargs)
+ self.result, exception, last_traceback, self.log = \
+ MPServer.run(self.id, self.func, self.args, self.kwargs)
except Exception, err:
- exception = err
- last_traceback = traceback.format_exc()
+ last_traceback = traceback.format_exc()
+ exception = (exception.__class__.__name__, unicode(str(err), 'utf8', 'replace'))
finally:
self.unlock()
self.emit(SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
- self.id, self.description, self.result, exception, last_traceback, self.log_dest.getvalue())
+ self.id, self.description, self.result, exception, last_traceback, self.log)
@@ -125,6 +122,8 @@ class JobManager(QAbstractTableModel):
self.cleanup = {}
self.device_job_icon = QVariant(QIcon(':/images/reader.svg'))
self.job_icon = QVariant(QIcon(':/images/jobs.svg'))
+ global MPServer
+ MPServer = Server()
def terminate_device_jobs(self):
changed = False
@@ -143,8 +142,8 @@ class JobManager(QAbstractTableModel):
try:
self.next_id += 1
job = job_class(self.next_id, description, lock, *args, **kwargs)
- QObject.connect(job, SIGNAL('finished()'), self.cleanup_jobs)
- QObject.connect(job, SIGNAL('status_update(int, int)'), self.status_update)
+ QObject.connect(job, SIGNAL('finished()'), self.cleanup_jobs, Qt.QueuedConnection)
+ QObject.connect(job, SIGNAL('status_update(int, int)'), self.status_update, Qt.QueuedConnection)
self.beginInsertRows(QModelIndex(), len(self.jobs), len(self.jobs))
self.jobs[self.next_id] = job
self.endInsertRows()
@@ -162,7 +161,7 @@ class JobManager(QAbstractTableModel):
def has_jobs(self):
return len(self.jobs.values()) > 0
- def run_conversion_job(self, slot, callable, *args, **kwargs):
+ def run_conversion_job(self, slot, callable, args=[], **kwargs):
'''
Run a conversion job.
@param slot: The function to call with the job result.
@@ -174,10 +173,10 @@ class JobManager(QAbstractTableModel):
job = self.create_job(ConversionJob, desc, self.conversion_lock,
callable, *args, **kwargs)
QObject.connect(job, SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
- self.job_done)
+ self.job_done, Qt.QueuedConnection)
if slot:
QObject.connect(job, SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
- slot)
+ slot, Qt.QueuedConnection)
priority = self.PRIORITY[str(QSettings().value('conversion job priority',
QVariant('Normal')).toString())]
job.start(priority)
@@ -195,10 +194,10 @@ class JobManager(QAbstractTableModel):
desc += kwargs.pop('job_extra_description', '')
job = self.create_job(DeviceJob, desc, self.device_lock, callable, *args, **kwargs)
QObject.connect(job, SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
- self.job_done)
+ self.job_done, Qt.QueuedConnection)
if slot:
QObject.connect(job, SIGNAL('jobdone(PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject, PyQt_PyObject)'),
- slot)
+ slot, Qt.QueuedConnection)
job.start()
return job.id
diff --git a/src/libprs500/gui2/main.py b/src/libprs500/gui2/main.py
index 24858a4503..98f0c08c2b 100644
--- a/src/libprs500/gui2/main.py
+++ b/src/libprs500/gui2/main.py
@@ -13,7 +13,6 @@
## with this program; if not, write to the Free Software Foundation, Inc.,
## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.Warning
import os, sys, textwrap, cStringIO, collections, traceback, shutil
-from functools import partial
from PyQt4.QtCore import Qt, SIGNAL, QObject, QCoreApplication, \
QSettings, QVariant, QSize, QThread
@@ -24,8 +23,6 @@ from PyQt4.QtSvg import QSvgRenderer
from libprs500 import __version__, __appname__, islinux, sanitize_file_name
from libprs500.ptempfile import PersistentTemporaryFile
from libprs500.ebooks.metadata.meta import get_metadata
-from libprs500.ebooks.lrf.web.convert_from import main as web2lrf
-from libprs500.ebooks.lrf.any.convert_from import main as _any2lrf
from libprs500.devices.errors import FreeSpaceError
from libprs500.devices.interface import Device
from libprs500.gui2 import APP_UID, warning_dialog, choose_files, error_dialog, \
@@ -56,7 +53,7 @@ from libprs500.ebooks.metadata.meta import set_metadata
from libprs500.ebooks.metadata import MetaInformation
from libprs500.ebooks import BOOK_EXTENSIONS
-any2lrf = partial(_any2lrf, gui_mode=True)
+
class Main(MainWindow, Ui_MainWindow):
@@ -557,7 +554,7 @@ class Main(MainWindow, Ui_MainWindow):
if data['password']:
args.extend(['--password', data['password']])
args.append(data['profile'])
- id = self.job_manager.run_conversion_job(self.news_fetched, web2lrf, args=args,
+ id = self.job_manager.run_conversion_job(self.news_fetched, 'web2lrf', args=[args],
job_description='Fetch news from '+data['title'])
self.conversion_jobs[id] = (pt, 'lrf')
self.status_bar.showMessage('Fetching news from '+data['title'], 2000)
@@ -606,9 +603,8 @@ class Main(MainWindow, Ui_MainWindow):
of.close()
cmdline.extend(['-o', of.name])
cmdline.append(pt.name)
-
id = self.job_manager.run_conversion_job(self.book_converted,
- any2lrf, args=cmdline,
+ 'any2lrf', args=[cmdline],
job_description='Convert book:'+d.title())
@@ -763,7 +759,7 @@ class Main(MainWindow, Ui_MainWindow):
_('There was a temporary error talking to the device. Please unplug and reconnect the device and or reboot.')).show()
return
print >>sys.stderr, 'Error in job:', description.encode('utf8')
- print >>sys.stderr, exception
+ print >>sys.stderr, exception[0], exception[1]
print >>sys.stderr, formatted_traceback.encode('utf8')
if not self.device_error_dialog.isVisible():
msg = u'
%s: '%(exception.__class__.__name__,) + unicode(str(exception), 'utf8', 'replace') + u'
' @@ -776,15 +772,17 @@ class Main(MainWindow, Ui_MainWindow): def conversion_job_exception(self, id, description, exception, formatted_traceback, log): print >>sys.stderr, 'Error in job:', description.encode('utf8') - print >>sys.stderr, log.encode('utf8') + if log: + print >>sys.stderr, log.encode('utf8') print >>sys.stderr, exception print >>sys.stderr, formatted_traceback.encode('utf8') - msg = u'%s: '%(exception.__class__.__name__,) + unicode(str(exception), 'utf8', 'replace') + u'
' + msg = u'%s: %s
'%exception msg += u'Failed to perform job: '+description msg += u'
Detailed traceback:
'
msg += formatted_traceback + ''
msg += 'Log:
'
- msg += log
+ if log:
+ msg += log
ConversionErrorDialog(self, 'Conversion Error', msg, show=True)
diff --git a/src/libprs500/gui2/news.py b/src/libprs500/gui2/news.py
index 7a59c7f198..85273de9fd 100644
--- a/src/libprs500/gui2/news.py
+++ b/src/libprs500/gui2/news.py
@@ -17,7 +17,7 @@ from PyQt4.QtGui import QMenu, QIcon, QDialog, QAction
from libprs500.gui2.dialogs.password import PasswordDialog
from libprs500.ebooks.lrf.web import builtin_profiles, available_profiles
-from libprs500.gui2.dialogs.user_profiles import UserProfiles
+from libprs500.ebooks.lrf.web.profiles import create_class
class NewsAction(QAction):
@@ -59,6 +59,9 @@ class NewsMenu(QMenu):
module = profile.title
username = password = None
fetch = True
+ if isinstance(profile, basestring):
+ module = profile
+ profile = create_class(module)
if profile.needs_subscription:
d = PasswordDialog(self, module + ' info dialog',
'Please enter your username and password for %s
If you do not have one, please subscribe to get access to the articles.
Click OK to proceed.'%(profile.title,))
@@ -68,7 +71,7 @@ class NewsMenu(QMenu):
else:
fetch = False
if fetch:
- data = dict(profile=profile, title=profile.title, username=username, password=password)
+ data = dict(profile=module, title=profile.title, username=username, password=password)
self.emit(SIGNAL('fetch_news(PyQt_PyObject)'), data)
def set_custom_feeds(self, feeds):
@@ -90,7 +93,7 @@ class CustomNewsMenu(QMenu):
self.connect(self, SIGNAL('triggered(QAction*)'), self.launch)
def launch(self, action):
- profile = UserProfiles.create_class(action.script)
+ profile = action.script
self.emit(SIGNAL('start_news_fetch(PyQt_PyObject, PyQt_PyObject)'),
profile, None)
diff --git a/src/libprs500/parallel.py b/src/libprs500/parallel.py
new file mode 100644
index 0000000000..1c2dd20ff3
--- /dev/null
+++ b/src/libprs500/parallel.py
@@ -0,0 +1,111 @@
+## Copyright (C) 2008 Kovid Goyal kovid@kovidgoyal.net
+## This program is free software; you can redistribute it and/or modify
+## it under the terms of the GNU General Public License as published by
+## the Free Software Foundation; either version 2 of the License, or
+## (at your option) any later version.
+##
+## This program is distributed in the hope that it will be useful,
+## but WITHOUT ANY WARRANTY; without even the implied warranty of
+## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+## GNU General Public License for more details.
+##
+## You should have received a copy of the GNU General Public License along
+## with this program; if not, write to the Free Software Foundation, Inc.,
+## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+'''
+Used to run jobs in parallel in separate processes.
+'''
+import re, sys, tempfile, os, subprocess, cPickle, cStringIO, traceback, atexit, time, binascii
+from functools import partial
+from libprs500.ebooks.lrf.any.convert_from import main as any2lrf
+from libprs500.ebooks.lrf.web.convert_from import main as web2lrf
+from libprs500 import iswindows
+
+PARALLEL_FUNCS = {
+ 'any2lrf' : partial(any2lrf, gui_mode=True),
+ 'web2lrf' : web2lrf,
+ }
+Popen = subprocess.Popen
+
+python = sys.executable
+if iswindows:
+ import win32con
+ Popen = partial(Popen, creationflags=win32con.CREATE_NO_WINDOW)
+ if hasattr(sys, 'frozen'):
+ python = os.path.join(os.path.dirname(python), 'parallel.exe')
+ else:
+ python = os.path.join(os.path.dirname(python), 'Scripts\\parallel.exe')
+
+def cleanup(tdir):
+ try:
+ import shutil
+ shutil.rmtree(tdir, True)
+ except:
+ pass
+
+class Server(object):
+
+ def __init__(self):
+ self.tdir = tempfile.mkdtemp('', 'libprs500_IPC_')
+ atexit.register(cleanup, self.tdir)
+ self.stdout = {}
+
+ def run(self, job_id, func, args=(), kwdargs={}):
+ job_id = str(job_id)
+ job_dir = os.path.join(self.tdir, job_id)
+ if os.path.exists(job_dir):
+ raise ValueError('Cannot run job. The job_id %s has already been used.')
+ os.mkdir(job_dir)
+ self.stdout[job_id] = cStringIO.StringIO()
+
+ job_data = os.path.join(job_dir, 'job_data.pickle')
+ cPickle.dump((func, args, kwdargs), open(job_data, 'wb'), -1)
+ prefix = ''
+ if hasattr(sys, 'frameworks_dir'):
+ fd = getattr(sys, 'frameworks_dir')
+ prefix = 'import sys; sys.frameworks_dir = "%s"; sys.frozen = "macosx_app"; '%fd
+ if fd not in os.environ['PATH']:
+ os.environ['PATH'] += ':'+fd
+ cmd = prefix + 'from libprs500.parallel import run_job; run_job(\'%s\')'%binascii.hexlify(job_data)
+
+ p = Popen((python, '-c', cmd), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ while p.returncode is None:
+ self.stdout[job_id].write(p.stdout.readline())
+ p.poll()
+ time.sleep(0.5) # Wait for half a second
+ self.stdout[job_id].write(p.stdout.read())
+
+ job_result = os.path.join(job_dir, 'job_result.pickle')
+ if not os.path.exists(job_result):
+ result, exception, traceback = None, ('ParallelRuntimeError', 'The worker process died unexpectedly.'), ''
+ else:
+ result, exception, traceback = cPickle.load(open(job_result, 'rb'))
+ log = self.stdout[job_id].getvalue()
+ self.stdout.pop(job_id)
+ return result, exception, traceback, log
+
+
+def run_job(job_data):
+ job_data = binascii.unhexlify(job_data)
+ job_result = os.path.join(os.path.dirname(job_data), 'job_result.pickle')
+ func, args, kwdargs = cPickle.load(open(job_data, 'rb'))
+ func = PARALLEL_FUNCS[func]
+ exception, tb = None, None
+ try:
+ result = func(*args, **kwdargs)
+ except (Exception, SystemExit), err:
+ result = None
+ exception = (err.__class__.__name__, unicode(str(err), 'utf-8', 'replace'))
+ tb = traceback.format_exc()
+
+ cPickle.dump((result, exception, tb), open(job_result, 'wb'))
+
+def main():
+ src = sys.argv[2]
+ job_data = re.search(r'run_job\(\'([a-f0-9A-F]+)\'\)', src).group(1)
+ run_job(job_data)
+
+ return 0
+
+if __name__ == '__main__':
+ sys.exit(main())
\ No newline at end of file