GitHub-Proxy/Cura
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/Ultimaker/Cura synced 2025-08-12 11:39:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'CURA-7492_Change_how_the_connection_status_is_determined'

Browse Source
This commit is contained in:
Ghostkeeper 2020-06-16 16:11:26 +02:00
commit 5286be77a1
No known key found for this signature in database
GPG Key ID: D2A8871EE34EC59A
2 changed files with 17 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
cura/API/ConnectionStatus.py
View File

@ -1,22 +1,21 @@
from typing import Optional from typing import Optional
from PyQt5.QtCore import QObject, pyqtSignal, QTimer, pyqtProperty from PyQt5.QtCore import QObject, pyqtSignal, pyqtProperty
from PyQt5.QtNetwork import QNetworkReply
from UM.TaskManagement.HttpRequestManager import HttpRequestManager from UM.TaskManagement.HttpRequestManager import HttpRequestManager
from cura.UltimakerCloud import UltimakerCloudConstants
class ConnectionStatus(QObject): class ConnectionStatus(QObject):
"""Status info for some web services""" """Provides an estimation of whether internet is reachable
UPDATE_INTERVAL = 10.0 # seconds Estimation is updated with every request through HttpRequestManager.
ULTIMAKER_CLOUD_STATUS_URL = UltimakerCloudConstants.CuraCloudAPIRoot + "/connect/v1/" Acts as a proxy to HttpRequestManager.internetReachableChanged without
exposing the HttpRequestManager in its entirety.
"""
__instance = None # type: Optional[ConnectionStatus] __instance = None # type: Optional[ConnectionStatus]
internetReachableChanged = pyqtSignal() internetReachableChanged = pyqtSignal()
umCloudReachableChanged = pyqtSignal()
@classmethod @classmethod
def getInstance(cls, *args, **kwargs) -> "ConnectionStatus": def getInstance(cls, *args, **kwargs) -> "ConnectionStatus":
@ -24,41 +23,19 @@ class ConnectionStatus(QObject):
cls.__instance = cls(*args, **kwargs) cls.__instance = cls(*args, **kwargs)
return cls.__instance return cls.__instance
def __init__(self, parent: Optional["QObject"] = None): def __init__(self, parent: Optional["QObject"] = None) -> None:
super().__init__(parent) super().__init__(parent)
self._http = HttpRequestManager.getInstance() manager = HttpRequestManager.getInstance()
self._statuses = { self._is_internet_reachable = manager.isInternetReachable # type: bool
self.ULTIMAKER_CLOUD_STATUS_URL: True, manager.internetReachableChanged.connect(self._onInternetReachableChanged)
"http://example.com": True
}
# Create a timer for automatic updates @pyqtProperty(bool, notify = internetReachableChanged)
self._update_timer = QTimer()
self._update_timer.setInterval(int(self.UPDATE_INTERVAL * 1000))
# The timer is restarted automatically
self._update_timer.setSingleShot(False)
self._update_timer.timeout.connect(self._update)
self._update_timer.start()
@pyqtProperty(bool, notify=internetReachableChanged)
def isInternetReachable(self) -> bool: def isInternetReachable(self) -> bool:
# Is any of the test urls reachable? return self._is_internet_reachable
return any(self._statuses.values())
def _update(self): def _onInternetReachableChanged(self, reachable: bool):
for url in self._statuses.keys(): if reachable != self._is_internet_reachable:
self._http.get( self._is_internet_reachable = reachable
url = url,
callback = self._statusCallback,
error_callback = self._statusCallback,
timeout = 5
)
def _statusCallback(self, reply: QNetworkReply, error: QNetworkReply.NetworkError = None):
url = reply.request().url().toString()
prev_statuses = self._statuses.copy()
self._statuses[url] = HttpRequestManager.replyIndicatesSuccess(reply, error)
if any(self._statuses.values()) != any(prev_statuses.values()):
self.internetReachableChanged.emit() self.internetReachableChanged.emit()

2
plugins/UM3NetworkPrinting/src/Cloud/CloudOutputDeviceManager.py
View File

@ -104,7 +104,7 @@ class CloudOutputDeviceManager:
self._api.getClusters(self._onGetRemoteClustersFinished, self._onGetRemoteClusterFailed) self._api.getClusters(self._onGetRemoteClustersFinished, self._onGetRemoteClusterFailed)
def _onGetRemoteClustersFinished(self, clusters: List[CloudClusterResponse]) -> None: def _onGetRemoteClustersFinished(self, clusters: List[CloudClusterResponse]) -> None:
"""Callback for when the request for getting the clusters is finished.""" """Callback for when the request for getting the clusters is successful and finished."""
self._um_cloud_printers = {m.getMetaDataEntry(self.META_CLUSTER_ID): m for m in self._um_cloud_printers = {m.getMetaDataEntry(self.META_CLUSTER_ID): m for m in
CuraApplication.getInstance().getContainerRegistry().findContainerStacks( CuraApplication.getInstance().getContainerRegistry().findContainerStacks(