I wrote a Python script for QGIS 3.36.2 (uses Python 3.12.3) that does the following:
- Create a layer
- Start an HTTP GET request to fetch new coordinates (runs asynchronously by default)
- Use these coordinates to draw a marker on the layer (the old marker is removed first, has to run on the main thread afaik)
Step 1 only happens once. 2. + 3. should run indefinitely but stop if there's an error or if the user stops the script. For testing I only want to run it e.g. 10 times.
What I've found/tried so far:
time.sleep()(as suggested here) freezes QGIS completely.schedscheduler (see code below) also blocks the main thread and freezes QGIS.threading.Timerwould start a new thread every time (and you wouldn't be able to stop the loop), so the answer advises against using it - untested because of that.- I can't use
Tkinterbecause QGIS' python doesn't support it. asyncio(as suggested here) doesn't seem to be fully supported in this QGIS version either (lots of errors when trying to run this example but it's working fine in the Python 3.9 console) and it's also kind of blocking because it uses coroutines (see this question; you canyield).
How do I repeat steps 2 and 3 multiple times if there's no error, e.g. 5 seconds after the last iteration finished, without blocking the GUI (especially the map viewer) with some type of sleep and preferably without using any extra libraries?
My code:
#imports here class ArrowDrawerClass: layer = None dataprovider = None feature = None repeat = True url = "someURL" repeatCounter = 0 myscheduler = sched.scheduler(time.time,time.sleep) def __init__(self): self.createNewLayer() def createNewLayer(self): layername = "ArrowLayer" self.layer = QgsVectorLayer('Point', layername, "memory") self.dataprovider = self.layer.dataProvider() self.feature = QgsFeature() #Set symbol, color,... of layer here QgsProject.instance().addMapLayers([self.layer]) def doRequest(self): request = QNetworkRequest(QUrl(self.url)) request.setTransferTimeout(10000) #10s self.manager = QNetworkAccessManager() self.manager.finished.connect(self.handleResponse) self.manager.get(request) def handleResponse(self, reply): err = reply.error() if err == QtNetwork.QNetworkReply.NetworkError.NoError: bytes = reply.readAll() replytext = str(bytes, 'utf-8').strip() #extract coordinates here ... self.drawArrow(x,y) else: self.displayError(str(err),reply.errorString()) def drawArrow(self,x,y): self.layer.dataProvider().truncate() #removes old marker point1 = QgsPointXY(x,y) self.feature.setGeometry(QgsGeometry.fromPointXY(point1)) self.dataprovider.addFeatures([self.feature]) self.layer.updateExtents() self.layer.triggerRepaint() self.repeatCounter += 1 self.repeatEverything() def displayError(self,code,msg): self.repeat = False #show error dialog here def start(self): self.myscheduler.enter(0,0,self.doRequest) self.myscheduler.run() def repeatEverything(self): print("counter:",self.repeatCounter) if self.repeat and self.repeatCounter<10: print("repeat") self.myscheduler.enter(5,0,self.test) #TODO: Call "self.doRequest()" instead self.myscheduler.run() else: print("don't repeat!") def test(self): print("test!") adc = ArrowDrawerClass() adc.start()