git.maemo.org Git - gc-dialer/blob - dialcentral/util/qore_utils.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import with_statement
   4
   5 import contextlib
   6 import logging
   7
   8 import qt_compat
   9 QtCore = qt_compat.QtCore
  10
  11 import misc
  12
  13
  14 _moduleLogger = logging.getLogger(__name__)
  15
  16
  17 class QThread44(QtCore.QThread):
  18         """
  19         This is to imitate QThread in Qt 4.4+ for when running on older version
  20         See http://labs.trolltech.com/blogs/2010/06/17/youre-doing-it-wrong
  21         (On Lucid I have Qt 4.7 and this is still an issue)
  22         """
  23
  24         def __init__(self, parent = None):
  25                 QtCore.QThread.__init__(self, parent)
  26
  27         def run(self):
  28                 self.exec_()
  29
  30
  31 class _WorkerThread(QtCore.QObject):
  32
  33         _taskComplete  = qt_compat.Signal(object)
  34
  35         def __init__(self, futureThread):
  36                 QtCore.QObject.__init__(self)
  37                 self._futureThread = futureThread
  38                 self._futureThread._addTask.connect(self._on_task_added)
  39                 self._taskComplete.connect(self._futureThread._on_task_complete)
  40
  41         @qt_compat.Slot(object)
  42         def _on_task_added(self, task):
  43                 self.__on_task_added(task)
  44
  45         @misc.log_exception(_moduleLogger)
  46         def __on_task_added(self, task):
  47                 if not self._futureThread._isRunning:
  48                         _moduleLogger.error("Dropping task")
  49
  50                 func, args, kwds, on_success, on_error = task
  51
  52                 try:
  53                         result = func(*args, **kwds)
  54                         isError = False
  55                 except Exception, e:
  56                         _moduleLogger.error("Error, passing it back to the main thread")
  57                         result = e
  58                         isError = True
  59
  60                 taskResult = on_success, on_error, isError, result
  61                 self._taskComplete.emit(taskResult)
  62
  63
  64 class FutureThread(QtCore.QObject):
  65
  66         _addTask = qt_compat.Signal(object)
  67
  68         def __init__(self):
  69                 QtCore.QObject.__init__(self)
  70                 self._thread = QThread44()
  71                 self._isRunning = False
  72                 self._worker = _WorkerThread(self)
  73                 self._worker.moveToThread(self._thread)
  74
  75         def start(self):
  76                 self._thread.start()
  77                 self._isRunning = True
  78
  79         def stop(self):
  80                 self._isRunning = False
  81                 self._thread.quit()
  82
  83         def add_task(self, func, args, kwds, on_success, on_error):
  84                 assert self._isRunning, "Task queue not started"
  85                 task = func, args, kwds, on_success, on_error
  86                 self._addTask.emit(task)
  87
  88         @qt_compat.Slot(object)
  89         def _on_task_complete(self, taskResult):
  90                 self.__on_task_complete(taskResult)
  91
  92         @misc.log_exception(_moduleLogger)
  93         def __on_task_complete(self, taskResult):
  94                 on_success, on_error, isError, result = taskResult
  95                 if not self._isRunning:
  96                         if isError:
  97                                 _moduleLogger.error("Masking: %s" % (result, ))
  98                         isError = True
  99                         result = StopIteration("Cancelling all callbacks")
 100                 callback = on_success if not isError else on_error
 101                 try:
 102                         callback(result)
 103                 except Exception:
 104                         _moduleLogger.exception("Callback errored")
 105
 106
 107 def create_single_column_list_model(columnName, **kwargs):
 108         """
 109         >>> class Single(object): pass
 110         >>> SingleListModel = create_single_column_list_model("s")
 111         >>> slm = SingleListModel([Single(), Single(), Single()])
 112         """
 113
 114         class SingleColumnListModel(QtCore.QAbstractListModel):
 115
 116                 def __init__(self, l = None):
 117                         QtCore.QAbstractListModel.__init__(self)
 118                         self._list = l if l is not None else []
 119                         self.setRoleNames({0: columnName})
 120
 121                 def __len__(self):
 122                         return len(self._list)
 123
 124                 def __getitem__(self, key):
 125                         return self._list[key]
 126
 127                 def __setitem__(self, key, value):
 128                         with scoped_model_reset(self):
 129                                 self._list[key] = value
 130
 131                 def __delitem__(self, key):
 132                         with scoped_model_reset(self):
 133                                 del self._list[key]
 134
 135                 def __iter__(self):
 136                         return iter(self._list)
 137
 138                 def __repr__(self):
 139                         return '<%s (%s)>' % (
 140                                 self.__class__.__name__,
 141                                 columnName,
 142                         )
 143
 144                 def rowCount(self, parent=QtCore.QModelIndex()):
 145                         return len(self._list)
 146
 147                 def data(self, index, role):
 148                         if index.isValid() and role == 0:
 149                                 return self._list[index.row()]
 150                         return None
 151
 152         if "name" in kwargs:
 153                 SingleColumnListModel.__name__ = kwargs["name"]
 154
 155         return SingleColumnListModel
 156
 157
 158 def create_tupled_list_model(*columnNames, **kwargs):
 159         """
 160         >>> class Column0(object): pass
 161         >>> class Column1(object): pass
 162         >>> class Column2(object): pass
 163         >>> MultiColumnedListModel = create_tupled_list_model("c0", "c1", "c2")
 164         >>> mclm = MultiColumnedListModel([(Column0(), Column1(), Column2())])
 165         """
 166
 167         class TupledListModel(QtCore.QAbstractListModel):
 168
 169                 def __init__(self, l = None):
 170                         QtCore.QAbstractListModel.__init__(self)
 171                         self._list = l if l is not None else []
 172                         self.setRoleNames(dict(enumerate(columnNames)))
 173
 174                 def __len__(self):
 175                         return len(self._list)
 176
 177                 def __getitem__(self, key):
 178                         return self._list[key]
 179
 180                 def __setitem__(self, key, value):
 181                         with scoped_model_reset(self):
 182                                 self._list[key] = value
 183
 184                 def __delitem__(self, key):
 185                         with scoped_model_reset(self):
 186                                 del self._list[key]
 187
 188                 def __iter__(self):
 189                         return iter(self._list)
 190
 191                 def __repr__(self):
 192                         return '<%s (%s)>' % (
 193                                 self.__class__.__name__,
 194                                 ', '.join(columnNames),
 195                         )
 196
 197                 def rowCount(self, parent=QtCore.QModelIndex()):
 198                         return len(self._list)
 199
 200                 def data(self, index, role):
 201                         if index.isValid() and 0 <= role and role < len(columnNames):
 202                                 return self._list[index.row()][role]
 203                         return None
 204
 205         if "name" in kwargs:
 206                 TupledListModel.__name__ = kwargs["name"]
 207
 208         return TupledListModel
 209
 210
 211 @contextlib.contextmanager
 212 def scoped_model_reset(model):
 213         model.beginResetModel()
 214         try:
 215                 yield
 216         finally:
 217                 model.endResetModel()
 218
 219
 220 def create_qobject(*classDef, **kwargs):
 221         """
 222         >>> Car = create_qobject(
 223         ...     ('model', str),
 224         ...     ('brand', str),
 225         ...     ('year', int),
 226         ...     ('inStock', bool),
 227         ...     name='Car'
 228         ... )
 229         >>> print Car
 230         <class '__main__.AutoQObject'>
 231         >>>
 232         >>> c = Car(model='Fiesta', brand='Ford', year=1337)
 233         >>> print c.model, c.brand, c.year, c.inStock
 234         Fiesta Ford 1337 False
 235         >>> print c
 236         <Car (model='Fiesta', brand='Ford', year=1337, inStock=False)>
 237         >>>
 238         >>> c.inStock = True
 239         >>>
 240         >>> print c.model, c.brand, c.year, c.inStock
 241         Fiesta Ford 1337 True
 242         >>> print c
 243         <Car (model='Fiesta', brand='Ford', year=1337, inStock=True)>
 244         """
 245
 246         class AutoQObject(QtCore.QObject):
 247
 248                 def __init__(self, **initKwargs):
 249                         QtCore.QObject.__init__(self)
 250                         for key, val in classDef:
 251                                 setattr(self, '_'+key, initKwargs.get(key, val()))
 252
 253                 def __repr__(self):
 254                         values = (
 255                                 '%s=%r' % (key, getattr(self, '_'+key))
 256                                 for key, value in classDef
 257                         )
 258                         return '<%s (%s)>' % (
 259                                 kwargs.get('name', self.__class__.__name__),
 260                                 ', '.join(values),
 261                         )
 262
 263                 for key, value in classDef:
 264                         nfy = locals()['_nfy_'+key] = qt_compat.Signal()
 265
 266                         def _get(key):
 267                                 def f(self):
 268                                         return self.__dict__['_'+key]
 269                                 return f
 270
 271                         def _set(key):
 272                                 def f(self, value):
 273                                         setattr(self, '_'+key, value)
 274                                         getattr(self, '_nfy_'+key).emit()
 275                                 return f
 276
 277                         setter = locals()['_set_'+key] = _set(key)
 278                         getter = locals()['_get_'+key] = _get(key)
 279
 280                         locals()[key] = qt_compat.Property(value, getter, setter, notify=nfy)
 281                 del nfy, _get, _set, getter, setter
 282
 283         return AutoQObject
 284
 285
 286 class QObjectProxy(object):
 287         """
 288         Proxy for accessing properties and slots as attributes
 289
 290         This class acts as a proxy for the object for which it is
 291         created, and makes property access more Pythonic while
 292         still allowing access to slots (as member functions).
 293
 294         Attribute names starting with '_' are not proxied.
 295         """
 296
 297         def __init__(self, rootObject):
 298                 self._rootObject = rootObject
 299                 m = self._rootObject.metaObject()
 300                 self._properties = [
 301                         m.property(i).name()
 302                         for i in xrange(m.propertyCount())
 303                 ]
 304
 305         def __getattr__(self, key):
 306                 value = self._rootObject.property(key)
 307
 308                 # No such property, so assume we call a slot
 309                 if value is None and key not in self._properties:
 310                         return getattr(self._rootObject, key)
 311
 312                 return value
 313
 314         def __setattr__(self, key, value):
 315                 if key.startswith('_'):
 316                         object.__setattr__(self, key, value)
 317                 else:
 318                         self._rootObject.setProperty(key, value)
 319
 320
 321 if __name__ == "__main__":
 322         import doctest
 323         print doctest.testmod()