alqtendpy's People
Forkers
epcankeralqtendpy's Issues
Tab bar corner widget style that can stretch
Autocompile *.ui on import
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
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
SIGINT handling (exit on ctrl+c)
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.
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.
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. ๐๐๐
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google โค๏ธ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.