altendky/alqtendpy

alqtendpy's Issues

Tab bar corner widget style that can stretch

https://gist.github.com/altendky/72d4b2582eda96c7f6da84abbfc7e1f6

Autocompile *.ui on import

altendky/basicpyqt5example@d981617#diff-bd949a6f240c11264122cca81a264bb8

class StringBuffer

https://gist.github.com/altendky/f7262e4f5aa5a96b0c5f9815d21021af

class StringBuffer(QObject):
    excrete = pyqtSignal(str)

    def __init__(self, parent=None, minimum=0.050):
        super(ConsoleBuffer, self).__init__(parent)
        self.minimum = minimum
        self.last_time = time.monotonic() - minimum
        self.buffer = []
        self.timer = QTimer()
        self.timer.setSingleShot(True)
        self.timer.timeout.connect(self._excrete)

    @pyqtSlot(str)
    def consume(self, s):
        self.buffer.append(s)

        delta = time.monotonic() - self.last_time
        remaining = self.minimum - delta
        if remaining <= 0:
            self._excrete()
        elif not self.timer.isActive():
            self.timer.start(int(1000 * remaining))

    def _excrete(self):
        self.timer.stop()
        s = ''.join(self.buffer)
        if len(s):
            self.last_time = time.monotonic()
            self.excrete.emit(s)
        self.buffer = []

process scheduler

https://gist.github.com/altendky/33edf26e14993a2cace153de8ab76de4

import collections
import sys

import attr
import PyQt5.QtCore
import PyQt5.QtWidgets


@attr.s
class ProcessRequest:
    qprocess = attr.ib()
    callback = attr.ib()

    def process_done(self, exit_code, exit_status):
        self.qprocess.finished.disconnect(self.process_done)

        self.callback(request=self)

    def start(self):
        self.qprocess.finished.connect(self.process_done)
        self.qprocess.start()


class ProcessSchedulerSignals(PyQt5.QtCore.QObject):
    process_finished = PyQt5.QtCore.pyqtSignal(PyQt5.QtCore.QProcess)
    queue_empty = PyQt5.QtCore.pyqtSignal()


@attr.s
class ProcessScheduler:
    limit = attr.ib()
    active = attr.ib(factory=list)
    queue = attr.ib(factory=collections.deque)
    signals = attr.ib(factory=ProcessSchedulerSignals)

    def submit_qprocess(self, qprocess):
        process_request = ProcessRequest(
            qprocess=qprocess,
            callback=self.process_done,
        )
        self.queue.append(process_request)
        self.maybe_start_another()
    
    def process_done(self, request):
        print('finished:', request.qprocess.program(), request.qprocess.arguments())
        self.active.remove(request)
        self.signals.process_finished.emit(request.qprocess)

        if len(self.queue) == 0 and len(self.active) == 0:
            self.signals.queue_empty.emit()

        self.maybe_start_another()

    def maybe_start_another(self):
        if len(self.queue) < 1:
            return

        if len(self.active) >= self.limit:
            return

        process_request = self.queue.popleft()
        self.active.append(process_request)
        process_request.start()
        print('started:', process_request.qprocess.program(), process_request.qprocess.arguments())


@attr.s
class Thingy:
    quit = attr.ib()
    finished = attr.ib(factory=list)
    scheduler = attr.ib(factory=lambda: ProcessScheduler(limit=20))

    def submit(self):
        self.scheduler.signals.process_finished.connect(self.process_finished)
        self.scheduler.signals.queue_empty.connect(self.queue_empty)

        for i in range(255):
            process = PyQt5.QtCore.QProcess()
            process.setProgram('ping')
            process.setArguments(['-c', '3', '192.168.0.{}'.format(i)])
            self.scheduler.submit_qprocess(qprocess=process)


    def process_finished(self, qprocess):
        self.finished.append(qprocess)
    
    def queue_empty(self):
        print('all processes completed')

        for process in self.finished:
            print(process.program(), process.arguments(), process.exitCode())
        
        self.quit()


def main():
    app = PyQt5.QtWidgets.QApplication(sys.argv)
    
    thingy = Thingy(quit=app.quit)
    PyQt5.QtCore.QTimer.singleShot(0, thingy.submit)

    return app.exec()


sys.exit(main())

WidgetDelegate

https://gist.github.com/altendky/b6aecb456be28abc09f01480d1135ca6
https://gist.github.com/altendky/ad82156d253c2dc8a8f2a278c798243a

class WidgetDelegate(QtWidgets.QStyledItemDelegate):
    def prepare(self, widget_factory, widget_filler):
        self.null_region = QtGui.QRegion()

        self.widget = widget_factory()
        self.widget.setup_ui()
        self.widget_filler = widget_filler
        self.widget.setAttribute(Qt.WA_DontShowOnScreen)

    def paint(self, painter, option, index):
        self.fill(rect=option.rect, index=index)

        if option.state & QtWidgets.QStyle.State_Selected:
            painter.fillRect(
                option.rect,
                option.palette.highlight(),
            )
        top_level = option.widget.window()
        target_offset = option.widget.mapTo(top_level, option.rect.topLeft())
        self.widget.render(
            painter,
            target_offset,
            self.null_region,
            QtWidgets.QWidget.DrawChildren,
        )

    def sizeHint(self, option, index):
        # we don't want this to be firing a lot so leave this in while changing things
        print('CustomDelegate.sizeHint()', time.time())
        rect = QRect(option.rect)
        # TODO: yuck...  find something better...  than forgetting stuff for awhile...
        width = option.widget.width()
        width -= 2 * option.widget.frameWidth()
        scroll_bar = option.widget.verticalScrollBar()
        if scroll_bar.isVisible():
            width -= scroll_bar.width()
        rect.setWidth(width)

        self.fill(rect=rect, index=index)

        return self.widget.sizeHint()

    def fill(self, rect, index):
        self.widget.show()
        self.widget.setGeometry(rect)
        self.widget_filler(self.widget, index=index)
        self.widget.close()

Wrapper for concurrent.futures.ThreadPoolExecutor

Or maybe this shouldn't exist because of QThreadPool.

https://gist.github.com/altendky/df3f7bdb41af4a4029a3b4597eba4fba

import concurrent.futures

import attr
from PyQt5 import Qt


class Executor(Qt.QObject):
    parent = attr.ib()
    max_workers = attr.ib(default=None)
    timer_period = attr.ib(default=0.5)
    complete_futures = attr.ib(factory=list)
    exception_futures = attr.ib(factory=list)
    _executor = attr.ib()
    _processing_futures = attr.ib(factory=list)
    _timer = attr.ib()                                 
    
    done = Qt.pyqtSignal()

    @_executor.default
    def _(self):
        return concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers)
    
    @_timer.default
    def _(self):
        timer = Qt.QTimer(self)
        timer.setInterval(self.timer_period * 1000)
        timer.timedout.connect(self._check)
        
        return timer
    
    def __attrs_post_init__(self):
        super().__init__(parent)
        
    def submit(self, f, *args, **kwargs):
        start_timer = len(self._processing_futures) == 0
        
        self._processing_futures.append(self._executor.submit(f, *args, **kwargs))

        if start_timer:
            self._timer.start()
    
    def _check(self):
        remaining = []
        
        for future in self._processing_futures:
            if future.running():
                remaining.append(future)
                continue
            
            append = self.complete_futures.append
            result = future.result
            
            try:
                append(result())
            except Exception as e:
                self.exception_futures.append(e)

        self._processing_futures = remaining
                
        if len(remaining) == 0:
            self._timer.stop()
            self.done.emit()

Qt logging message handler

https://github.com/altendky/stlib/blob/5224e6bf2ae6ca025e46c31dbb62ee430c24c9cf/epyqlib/utils/qt.py#L133-L149

QtCore.qInstallMessageHandler(epyqlib.utils.qt.message_handler)

https://github.com/altendky/st/blob/1ea4dbf6cc92bfc16dd33db6388f3e1ca8ae596d/epyq/__main__.py#L347

# http://stackoverflow.com/a/35902894/228539
def message_handler(mode, context, message):
    mode_strings = {
        QtCore.QtInfoMsg: 'INFO',
        QtCore.QtWarningMsg: 'WARNING',
        QtCore.QtCriticalMsg: 'CRITICAL',
        QtCore.QtFatalMsg: 'FATAL'
    }

    mode = mode_strings.get(mode, 'DEBUG')

    print('qt_message_handler: f:{file} l:{line} f():{function}'.format(
        file=context.file,
        line=context.line,
        function=context.function
    ))
    print('  {}: {}\n'.format(mode, message))

But! Spit it out through standard Python logging, not print.

Add timeout to async signal wrappers

@attr.s
class DeferredForSignal:
    signal = attr.ib()
    deferred = attr.ib(
        default=attr.Factory(
            factory=lambda self: twisted.internet.defer.Deferred(
                canceller=self.cancelled,
            ),
            takes_self=True,
        ),
    )
    timeout_call = attr.ib(default=None)

    def connect(self, timeout=None):
        self.signal.connect(self.slot)

        if timeout is not None:
            import twisted.internet.reactor
            self.timeout_call = twisted.internet.reactor.callLater(
                timeout,
                self.time_out,
            )

    def time_out(self):
        self.signal.disconnect(self.slot)
        self.deferred.errback(epyqlib.utils.twisted.RequestTimeoutError())

    def disconnect(self):
        self.signal.disconnect(self.slot)
        if self.timeout_call is not None:
            self.timeout_call.cancel()

    def cancelled(self, deferred):
        self.disconnect()

    def slot(self, *args):
        self.disconnect()
        self.deferred.callback(args)


def signal_as_deferred(signal, timeout=None, f=None, *args, **kwargs):
    dfs = DeferredForSignal(signal=signal)
    dfs.connect(timeout=timeout)

    if f is not None:
        f(*args, **kwargs)

    return dfs.deferred

Ease adding class' path to one of its attributes

https://repl.it/@altendky/options-to-set-identifier-path-on-class

Such as for

https://github.com/altendky/stlib/blob/874e741bb3eb2c9ff7af15a4918af933cce8eb2c/epyqlib/utils/qt.py#L1138-L1176

SIGINT handling (exit on ctrl+c)

https://github.com/altendky/stlib/blob/3aa77bbab57b40555f364478c0e94f8f69ec0aac/epyqlib/utils/qt.py#L1125-L1137

def sigint_handler(signal_number, stack_frame):
    QtWidgets.QApplication.exit(128 + signal_number)


def setup_sigint():
    signal.signal(signal.SIGINT, sigint_handler)

    # Regularly give Python a chance to receive signals such as ctrl+c
    timer = QtCore.QTimer()
    timer.start(200)
    timer.timeout.connect(lambda: None)

    return timer

Make a app context manager? Or at least understand the situation better.

https://www.riverbankcomputing.com/pipermail/pyqt/2019-March/041489.html

sys.excepthook dialog

combine with altendky/altendpy#7

class AspectRatioPadding

https://gist.github.com/e46dcc26bad9dc462ee55e5b959ffa29

class AspectRatioPadding(Qt.QWidget):
    def __init__(self, child, parent=None):
        super().__init__(parent)

        self.child = child
        self.child.setParent(self)

    def resizeEvent(self, event):
        new_size = event.size()
        child_size = self.child.sizeHint().scaled(
            new_size,
            Qt.Qt.KeepAspectRatio,
        )

        offset = (new_size - child_size) / 2

        self.child.setGeometry(
            offset.width(),
            offset.height(),
            child_size.width(),
            child_size.height(),
        )

    def sizeHint(self):
        return self.child.sizeHint()

Support Python async syntax for signals

A related reference.

qutebrowser/qutebrowser#3327 (comment)