Created
December 9, 2024 04:18
-
-
Save minimalefforttech/0969936ccad3c55237bc1799ad18e354 to your computer and use it in GitHub Desktop.
An example UI to demonstrate delegates, remote networks and data mapping
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
This example demonstrates aims to deomonstrate three simple concepts: | |
1. How we can retrieve data from a remote network asyncronously | |
1a. How we can load and display images from a remote network asynchronously | |
2. How to display remote images inside a model view | |
3. How to use data model mapping to edit a model | |
I have not added any "prettty visuals" or error checking. | |
This is purely to get you started with the concepts, it is on you to take this further depending on your own requirements. | |
CC-0, While I appreciate the links you do not have to credit me in any usage. | |
Thanks, | |
Alex Telford | |
https://minimaleffort.tech | |
""" | |
from dataclasses import dataclass | |
from weakref import proxy | |
import json | |
import urllib | |
from enum import Enum, IntEnum, auto | |
import platform | |
try: | |
from PySide6 import QtCore, QtGui, QtWidgets, QtNetwork | |
except ImportError: | |
from PySide2 import QtCore, QtGui, QtWidgets, QtNetwork | |
ID_ROLE = QtCore.Qt.UserRole + 1 | |
class Color(IntEnum): | |
""" Basic enum for our color options, this is so we can pass ints directly to the combobox index. | |
While you can map a combobox text you need to add an intermediary. | |
""" | |
black = 0 | |
white = auto() | |
red = auto() | |
green = auto() | |
blue = auto() | |
@dataclass | |
class ItemData: | |
""" Simple container to store the data associated with each row in the model | |
""" | |
id:str | |
text:str | |
font_size:int=16 | |
font_color:Color=Color.black | |
thumbnail:QtGui.QImage=None | |
class ThumbnailDelegate(QtWidgets.QStyledItemDelegate): | |
""" This delegate is used to draw either a "loading..." text or the image once loaded. That's all | |
You could expand on this to also provide the size hints or loading spinners, etc | |
""" | |
def paint(self, painter:QtGui.QPainter, option, index): | |
image = index.data(QtCore.Qt.DisplayRole) | |
painter.save() | |
if not image: | |
# Draw loading text, could do a spinner here | |
# Draw the text | |
QtWidgets.QApplication.style().drawItemText( | |
painter, | |
option.rect, | |
QtCore.Qt.AlignVCenter|QtCore.Qt.AlignLeft, | |
QtGui.QColor(QtCore.Qt.white), | |
True, | |
"Loading...") | |
else: | |
painter.drawPixmap(option.rect, QtGui.QPixmap.fromImage(image).scaledToHeight(option.rect.height())) | |
painter.restore() | |
class CatProvider(QtCore.QObject): | |
""" The network operations around CATAAS (Cat as a service) | |
Where possible it is a good idea to separate your queries/data/views | |
""" | |
def __init__(self, parent:QtCore.QObject=None): | |
super().__init__(parent) | |
self._base_url = QtCore.QUrl("https://cataas.com/") | |
self._api_endpoint = "/api/cats" | |
self._thumbnail_endpoint = "/cat/{id}" | |
self._generator_endpoint = "/cat/{id}/says/{text}" | |
self._network_manager = QtNetwork.QNetworkAccessManager() | |
def request_cat_list(self, start:int=0, limit:int=100)->QtNetwork.QNetworkReply: | |
url = QtCore.QUrl(self._base_url) | |
url.setPath(self._api_endpoint) | |
query = QtCore.QUrlQuery() | |
# This just turns the url into https://some_url.com/api/endpoint?param=value | |
query.addQueryItem("skip", str(start)) | |
query.addQueryItem("limit", str(limit)) | |
url.setQuery(query) | |
# Create the network request, this can be a simple url like here or contain auth tokens, json data, etc. | |
# cataas has none of that so we just use it directly | |
request = QtNetwork.QNetworkRequest() | |
request.setUrl(url) | |
reply = self._network_manager.get(request) | |
return reply | |
def request_cat_thumbnail(self, cat_id:str, size:QtCore.QSize=QtCore.QSize(64, 64))->QtNetwork.QNetworkReply: | |
""" Request a thumbnail """ | |
url = QtCore.QUrl(self._base_url) | |
url.setPath(self._thumbnail_endpoint.format(id=cat_id)) | |
query = QtCore.QUrlQuery() | |
query.addQueryItem("width", str(size.width())) | |
query.addQueryItem("height", str(size.height())) | |
url.setQuery(query) | |
request = QtNetwork.QNetworkRequest() | |
request.setUrl(url) | |
reply = self._network_manager.get(request) | |
reply.setProperty("cat_id", cat_id) | |
return reply | |
def request_cat_quote(self, cat_id:str, text:str=None, font_size:int=16, font_color:Color=Color.black)->QtNetwork.QNetworkReply: | |
""" Generate a new image """ | |
if isinstance(font_color, (str, bytes, QtCore.QByteArray)): | |
font_color:str = getattr(Color, font_color).value | |
else: | |
font_color:str = Color(font_color).value | |
url = QtCore.QUrl(self._base_url) | |
query = QtCore.QUrlQuery() | |
query.addQueryItem("type", "square") | |
if text: | |
text = urllib.parse.quote(text) | |
url.setPath(self._generator_endpoint.format(id=cat_id, text=text)) | |
query.addQueryItem("fontSize", str(font_size)) | |
query.addQueryItem("fontColor", str(font_color)) | |
else: | |
url.setPath(self._thumbnail_endpoint.format(id=cat_id)) | |
request = QtNetwork.QNetworkRequest() | |
request.setUrl(url) | |
reply = self._network_manager.get(request) | |
reply.setProperty("cat_id", cat_id) | |
return reply | |
class AsyncCatModel(QtCore.QAbstractTableModel): | |
""" A basic item model that provides the cat entries and stores the data. | |
Ideally this would be split into a data model and a query provider, but for example purposes I have merged them. | |
""" | |
# Optional, you can use this to show a loading spinner elsewhere | |
loading_changed = QtCore.Signal(bool) | |
def __init__(self, provider:CatProvider, parent=None): | |
super().__init__(parent) | |
self._provider = proxy(provider) # The model does not "own" this, it is more of a global state | |
self._headers = ["Thumbnail", "Text", "Font Size", "Font Color"] | |
self._items = [] | |
self._limit = 100 # Or whatever | |
self._batchsize = 20 | |
self._loading = False | |
self._query_reply:QtNetwork.QNetworkReply = None # This is the reply for querying cat data | |
self._thumbnail_replies = [] # This is the reply for querying cat thumbnails | |
def is_loading(self)->bool: | |
return self._loading | |
loading = QtCore.Property(bool, is_loading, notify=loading_changed) | |
def headerData(self, section:int, orientation, role=QtCore.Qt.DisplayRole): | |
if orientation == QtCore.Qt.Horizontal and role in (QtCore.Qt.DisplayRole, QtCore.Qt.EditRole): | |
return self._headers[section] # Should check range | |
return None | |
def rowCount(self, parent=QtCore.QModelIndex()): | |
return len(self._items) | |
def columnCount(self, parent=QtCore.QModelIndex()): | |
return len(self._headers) | |
def data(self, index:QtCore.QModelIndex, role): | |
""" Convert the data from the structure to the appropriate representation. | |
Ideally this should just be direct data with minor conversions (color int/text) with delegates on top | |
But realistically we often just bung it all in here. | |
Try not to have much computation in here as this gets called a lot. | |
""" | |
row = index.row() | |
column = index.column() | |
if row < 0 or row >= len(self._items): | |
return None | |
if role == ID_ROLE: | |
return self._items[row].id | |
elif role == QtCore.Qt.SizeHintRole: | |
if column == 0: | |
if self._items[row].thumbnail: | |
return self._items[row].thumbnail.size() | |
return QtCore.QSize(100, 100) | |
elif role in (QtCore.Qt.DisplayRole, QtCore.Qt.EditRole): | |
if column == 0: | |
return self._items[row].thumbnail | |
elif column == 1: | |
return self._items[row].text | |
elif column == 2: | |
return self._items[row].font_size | |
elif column == 3: | |
# data mappers use EditRole, but we want to see the text | |
if role == QtCore.Qt.DisplayRole: | |
return self._items[row].font_color.name | |
return int(self._items[row].font_color.value) | |
elif role == QtCore.Qt.BackgroundRole: | |
# The color column should be a delegate too, but again, this is an example | |
if column == 3: | |
return QtGui.QColor(self._items[row].font_color.name) | |
elif role == QtCore.Qt.ForegroundRole: | |
if column == 3: | |
if self._items[row].font_color == Color.black: | |
return QtGui.QColor(QtCore.Qt.white) | |
return QtGui.QColor(QtCore.Qt.black) | |
return None | |
def setData(self, index, value, role=QtCore.Qt.EditRole): | |
""" Again, just maps the value to the appropriate entry in the structure | |
""" | |
if role != QtCore.Qt.EditRole: | |
return False | |
row = index.row() | |
column = index.column() | |
if row < 0 or row >= len(self._items): | |
return None | |
if column == 0: | |
pass | |
elif column == 1: | |
self._items[row].text = str(value or "") | |
self.dataChanged.emit(index, index, [role]) | |
elif column == 2: | |
self._items[row].font_size = int(value) | |
self.dataChanged.emit(index, index, [role]) | |
elif column == 3: | |
self._items[row].font_color = Color(value) | |
self.dataChanged.emit(index, index, [role]) | |
return True | |
@QtCore.Slot() | |
def reload(self): | |
""" If you do not call fetchMore once it has cleared the model will remain empty | |
""" | |
self.reset() | |
self.fetchMore(QtCore.QModelIndex()) | |
@QtCore.Slot() | |
def reset(self): | |
""" Reset the model, this is a convenience way to reset any internals to a clean state. | |
""" | |
self.beginResetModel() | |
self._items = [] | |
if self._query_reply: # can't stop this but at least don't listen to the result | |
self._query_reply.readyRead.disconnect(self._reply_ready) | |
for each in self._thumbnail_replies: | |
each.readyRead.disconnect(self._thumbnail_ready) | |
self._thumbnail_replies = [] | |
self._query_reply = None | |
self._thumbnail_reply = None | |
self._loading = False | |
self.loading_changed.emit(True) | |
self.endResetModel() | |
def canFetchMore(self, parent=QtCore.QModelIndex())->bool: | |
""" This is called as a user is scrolling to see if there are any more items to be loaded. | |
""" | |
if self._loading: | |
return False | |
if len(self._items) >= self._limit: | |
return False | |
return True | |
def fetchMore(self, parent=QtCore.QModelIndex()): | |
""" This is called as a user is scrolling to fetch the next entries from the server | |
""" | |
if self._loading: | |
return False | |
self._query_reply = self._provider.request_cat_list(start=len(self._items), limit=self._limit) | |
self._loading = True | |
self.loading_changed.emit(True) | |
self._query_reply.readyRead.connect(self._query_reply_ready) # Can also connect network.finished instead | |
# When working with network data you also need to handle errors and malformatted responses, I've left this blank for the demo | |
# reply.errorOccurred.connect(self._on_error) | |
@QtCore.Slot() | |
def _query_reply_ready(self): | |
# Ideally here you should validate the data is complete, sometimes cataas returns a partial response | |
data = self._query_reply.readAll().toStdString() | |
try: | |
items = json.loads(data) | |
except: | |
print(data) | |
raise | |
count = len(self._items) | |
for each in items: | |
if each["mimetype"] != "image/jpeg": | |
continue # Only doing images right now, the catapi also returns gifs which you could prefilter if you wanted | |
self._items.append(ItemData( | |
id=each["_id"], | |
text="", | |
)) | |
reply = self._provider.request_cat_thumbnail(each["_id"], QtCore.QSize(64, 64)) | |
self._thumbnail_replies.append(reply) | |
reply.readyRead.connect(self._thumbnail_reply_ready) | |
# Again, check for errors... | |
# Technically this goes before we append, but it's python so this works fine | |
self.beginInsertRows(QtCore.QModelIndex(), count, len(self._items)) | |
self.endInsertRows() | |
@QtCore.Slot() | |
def _thumbnail_reply_ready(self): | |
# By calling QObject.sender we can see which reply triggered this connection, saves us keeping track. | |
# You can also connect straight to the QNetworkAccessManager to get it in the arguments. | |
reply = self.sender() | |
id_ = reply.property("cat_id") | |
for i, each in enumerate(self._items): | |
if each.id == id_: | |
break | |
else: | |
return # Probably no longer available | |
# This will crash if the image is not an image, again: do your own validation | |
image = QtGui.QImage.fromData(reply.readAll()) | |
each.thumbnail = image | |
index = self.index(i, 0) | |
self.dataChanged.emit(index, index, [QtCore.Qt.DisplayRole]) | |
if reply in self._thumbnail_replies: | |
self._thumbnail_replies.remove(reply) | |
class MappedForm(QtWidgets.QWidget): | |
""" Simple container to demonstrate mapping data from a model to a separate set of inputs | |
Also shows doing async image queries again, because why not. | |
""" | |
def __init__(self, provider:CatProvider, model:QtCore.QAbstractItemModel, parent=None): | |
super().__init__(parent) | |
self._reply:QtNetwork.QNetworkReply = None | |
self._provider = proxy(provider) # The form does not "own" this, it is more of a global state | |
layout = QtWidgets.QFormLayout(self) | |
layout.setFieldGrowthPolicy(QtWidgets.QFormLayout.AllNonFixedFieldsGrow) | |
self._image = QtWidgets.QLabel("Select an item") | |
self._image.setFixedSize(QtCore.QSize(256, 256)) | |
layout.addRow(self._image) | |
self._text_input = QtWidgets.QLineEdit() | |
layout.addRow("Text", self._text_input) | |
self._font_size_input = QtWidgets.QSpinBox() | |
self._font_size_input.setRange(6, 64) | |
layout.addRow("Font Size", self._font_size_input) | |
self._color_input = QtWidgets.QComboBox() | |
self._color_input.addItems([c.name for c in Color]) | |
layout.addRow("Color", self._color_input) | |
self._mapper = QtWidgets.QDataWidgetMapper(self) | |
self._mapper.setModel(model) | |
self._mapper.setOrientation(QtCore.Qt.Horizontal) | |
# The index here is the column, ideally store these somewhere rather than hard coded | |
self._mapper.addMapping(self._text_input, 1) | |
self._mapper.addMapping(self._font_size_input, 2) | |
# You can specify a property name, useful for mapping against custom editors. | |
self._mapper.addMapping(self._color_input, 3, b"currentIndex") | |
if platform.system() == "Darwin": | |
# comboboxes don't get focus on macos | |
self._color_input.currentIndexChanged.connect(self._mapper.submit) | |
self._color_input.currentIndexChanged.connect(self._invalidate_preview) | |
self._update_timer = QtCore.QTimer(self) | |
self._update_timer.setSingleShot(True) | |
self._update_timer.setInterval(200) # Wait before pinging server in case user is just clicking through stuff. | |
self._update_timer.timeout.connect(self._update_preview) | |
self._text_input.textEdited.connect(self._invalidate_preview) | |
self._font_size_input.valueChanged.connect(self._invalidate_preview) | |
def set_index(self, index): | |
self._mapper.setCurrentIndex(index.row()) | |
self._invalidate_preview() | |
def _invalidate_preview(self): | |
""" This is triggered when any of the data is changed, You could put a spinner over the image rather than replacing or whatever you want | |
""" | |
if self._reply: | |
self._reply.readyRead.disconnect(self._preview_ready) | |
self._reply = None | |
self._image.clear() | |
self._image.setText("Loading...") | |
self._update_timer.start() | |
def _update_preview(self): | |
""" Data is ready to update, send a request to the server for the composited image | |
""" | |
model = self._mapper.model() | |
index = model.index(self._mapper.currentIndex(), 0) | |
cat_id = model.data(index, ID_ROLE) | |
# Should probably get these from the model... | |
font_size = self._font_size_input.value() | |
font_color = self._color_input.currentText() | |
text = self._text_input.text() | |
self._reply = self._provider.request_cat_quote(cat_id, text, font_size, font_color) | |
self._reply.readyRead.connect(self._preview_ready) # Can also connect network.finished instead | |
@QtCore.Slot() | |
def _preview_ready(self): | |
""" The server has generated the image, you can display it now | |
""" | |
if not self._reply: | |
return | |
image = QtGui.QImage.fromData(self._reply.readAll()) # Can also load direct to QPixmap | |
self._image.clear() | |
self._image.setPixmap(QtGui.QPixmap.fromImage(image)) | |
self._image.setScaledContents(True) | |
self._reply = None | |
class Widget(QtWidgets.QWidget): | |
""" A simple widget showing networking, mapping and async images | |
""" | |
def __init__(self, parent=None): | |
super().__init__(parent) | |
self.setWindowTitle("Network and Mapping Example") | |
provider = CatProvider(self) | |
layout = QtWidgets.QHBoxLayout(self) | |
view = QtWidgets.QTreeView() | |
view.setItemDelegateForColumn(0, ThumbnailDelegate(self)) | |
layout.addWidget(view) | |
layout.setStretchFactor(view, 1) | |
model = AsyncCatModel(provider=provider, parent=self) | |
view.setModel(model) | |
right_panel = MappedForm(provider=provider, model=model) | |
layout.addWidget(right_panel) | |
view.clicked.connect(right_panel.set_index) | |
model.reload() | |
widget = Widget() | |
widget.show() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment