ollama Interface: comparison OllamaInterface/OllamaClient.py

-:eb1dec15b2f0
+:3118d16e526e
 @signal replyReceived(content:str, role:str, done:bool) emitted after a response
 from the 'ollama' server was received
 @signal modelsList(modelNames:list[str]) emitted after the list of model
 names was obtained from the 'ollama' server
-@signal detailedModelsList(models:list[dict]) emitted after the list of
-models was obtained from the 'ollama' server giving some model details
-@signal runningModelsList(models:list[dict]) emitted after the list of
-running models was obtained from the 'ollama' server giving some model
-execution details
 @signal pullStatus(msg:str, id:str, total:int, completed:int) emitted to indicate
 the status of a pull request as reported by the 'ollama' server
 @signal serverVersion(version:str) emitted after the server version was obtained
 from the 'ollama' server
 @signal finished() emitted to indicate the completion of a request
 responsiveness
 """
 replyReceived = pyqtSignal(str, str, bool)
 modelsList = pyqtSignal(list)
-detailedModelsList = pyqtSignal(list)
-runningModelsList = pyqtSignal(list)
 pullStatus = pyqtSignal(str, str, int, int)
 serverVersion = pyqtSignal(str)
 finished = pyqtSignal()
 errorOccurred = pyqtSignal(str)
 serverStateChanged = pyqtSignal(bool)
 def listDetails(self):
 """
 Public method to request a list of models available locally from the 'ollama'
 server with some model details.
 """
-# TODO: not implemented yet
+response = self.__sendSyncRequest("tags")
-self.__sendRequest("tags", processResponse=self.__processDetailedModelsList)
-def __processDetailedModelsList(self, response):
-"""
-Private method to process the tags response of the 'ollama' server extracting
-some model details.
-@param response dictionary containing the tags response
-@type dict
-"""
 models = []
-with contextlib.suppress(KeyError):
+if response is not None:
-for model in response["models"]:
+with contextlib.suppress(KeyError):
-name = model["name"]
+for model in response["models"]:
-if name:
+name = model["name"]
-models.append(
+if name:
-{
+models.append(
-"name": name,
+{
-"id": model["digest"][:20],  # first 20 characters only
+"name": name,
-"size": model["size"],
+"id": model["digest"][:20],  # first 20 characters only
-"modified": datetime.datetime.fromisoformat(
+"size": model["size"],
-model["modified_at"]
+"modified": datetime.datetime.fromisoformat(
-),
+model["modified_at"]
-}
+),
-)
+}
-self.detailedModelsList.emit(models)
+)
+return models
 def listRunning(self):
 """
 Public method to request a list of running models from the 'ollama' server.
 """
-# TODO: not implemented yet
+response = self.__sendSyncRequest("ps")
-self.__sendRequest("ps", processResponse=self.__processRunningModelsList)
-def __processRunningModelsList(self, response):
-"""
-Private method to process the ps response of the 'ollama' server extracting
-some model execution details.
-@param response dictionary containing the ps response
-@type dict
-"""
 models = []
-with contextlib.suppress(KeyError):
+if response is not None:
-for model in response["models"]:
+with contextlib.suppress(KeyError):
-name = model["name"]
+for model in response["models"]:
-if name:
+name = model["name"]
-if model["size_vram"] == 0:
+if name:
-processor = self.tr("100% CPU")
+if model["size_vram"] == 0:
-elif model["size_vram"] == model["size"]:
+processor = self.tr("100% CPU")
-processor = self.tr("100% GPU")
+elif model["size_vram"] == model["size"]:
-elif model["size_vram"] > model["size_"] or model["size"] == 0:
+processor = self.tr("100% GPU")
-processor = self.tr("unknown")
+elif model["size_vram"] > model["size"] or model["size"] == 0:
-else:
+processor = self.tr("unknown")
-sizeCpu = model["size"] - model["size_vram"]
+else:
-cpuPercent = round(sizeCpu / model["size_vram"] * 100)
+sizeCpu = model["size"] - model["size_vram"]
-processor = self.tr("{0}% / {1}% CPU / GPU").format(
+cpuPercent = round(sizeCpu / model["size_vram"] * 100)
-cpuPercent, 100 - cpuPercent
+processor = self.tr("{0}% / {1}% CPU / GPU").format(
+cpuPercent, 100 - cpuPercent
+)
+models.append(
+{
+"name": name,
+"id": model["digest"][:20],  # first 20 characters only
+"size": model["size"],
+"size_vram": model["size_vram"],
+"processor": processor,
+"expires": datetime.datetime.fromisoformat(
+model["expires_at"]
+),
+}
 )
-models.append(
-{
+return models
-"name": name,
-"id": model["digest"][:20],  # first 20 characters only
-"size": model["size"],
-"size_vram": model["size_vram"],
-"processor": processor,
-"expires": datetime.datetime.fromisoformat(
-model["expires_at"]
-),
-}
-)
-self.runningModelsList.emit(models)
 def version(self):
 """
 Public method to request the version from the 'ollama' server.
 """
 @return current client state
 @rtype OllamaClientState
 """
 return self.__state
-def __sendRequest(self, endpoint, data=None, processResponse=None):
+def __getServerReply(self, endpoint, data=None):
 """
-Private method to send a request to the 'ollama' server and handle its
+Private method to send a request to the 'ollama' server and return a reply
-responses.
+object.
 @param endpoint 'ollama' API endpoint to be contacted
 @type str
 @param data dictionary containing the data to send to the server
 (defaults to None)
 @type dict (optional)
-@param processResponse function handling the received data (defaults to None)
+@return 'ollama' server reply
-@type function (optional)
+@rtype QNetworkReply
 """
-self.__state = OllamaClientState.Requesting
+ollamaUrl =  QUrl(
-ollamaUrl = QUrl(
 "{0}://{1}:{2}/api/{3}".format(
 self.__plugin.getPreferences("OllamaScheme"),
 (
 "127.0.0.1"
 if self.__localServer
 )
 jsonData = json.dumps(data).encode("utf-8")
 reply = self.__networkManager.post(request, jsonData)
 else:
 reply = self.__networkManager.get(request)
+reply.errorOccurred.connect(lambda error: self.__errorOccurred(error, reply))
+return reply
+def __sendRequest(self, endpoint, data=None, processResponse=None):
+"""
+Private method to send a request to the 'ollama' server and handle its
+responses.
+@param endpoint 'ollama' API endpoint to be contacted
+@type str
+@param data dictionary containing the data to send to the server
+(defaults to None)
+@type dict (optional)
+@param processResponse function handling the received data (defaults to None)
+@type function (optional)
+"""
+self.__state = OllamaClientState.Requesting
+##ollamaUrl = QUrl(
+##"{0}://{1}:{2}/api/{3}".format(
+##self.__plugin.getPreferences("OllamaScheme"),
+##(
+##"127.0.0.1"
+##if self.__localServer
+##else self.__plugin.getPreferences("OllamaHost")
+##),
+##(
+##self.__plugin.getPreferences("OllamaLocalPort")
+##if self.__localServer
+##else self.__plugin.getPreferences("OllamaPort")
+##),
+##endpoint,
+##)
+##)
+##request = QNetworkRequest(ollamaUrl)
+##if data is not None:
+##request.setHeader(
+##QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json"
+##)
+##jsonData = json.dumps(data).encode("utf-8")
+##reply = self.__networkManager.post(request, jsonData)
+##else:
+##reply = self.__networkManager.get(request)
+##
+reply = self.__getServerReply(endpoint=endpoint, data=data)
 reply.finished.connect(lambda: self.__replyFinished(reply))
-reply.errorOccurred.connect(lambda error: self.__errorOccurred(error, reply))
+##reply.errorOccurred.connect(lambda error: self.__errorOccurred(error, reply))
 reply.readyRead.connect(lambda: self.__processData(reply, processResponse))
 self.__replies.append(reply)
 def __replyFinished(self, reply):
 """
 if buffer:
 with contextlib.suppress(json.JSONDecodeError):
 data = json.loads(buffer)
 if data and processResponse:
 processResponse(data)
+def __sendSyncRequest(self, endpoint, data=None):
+"""
+Private method to send a request to the 'ollama' server and handle its
+responses.
+@param endpoint 'ollama' API endpoint to be contacted
+@type str
+@param data dictionary containing the data to send to the server
+(defaults to None)
+@type dict (optional)
+"""
+self.__state = OllamaClientState.Requesting
+reply = self.__getServerReply(endpoint=endpoint, data=data)
+while not reply.isFinished():
+QCoreApplication.processEvents()
+QThread.msleep(100)
+reply.deleteLater()
+self.__state = OllamaClientState.Finished
+if reply.error() == QNetworkReply.NetworkError.NoError:
+buffer = bytes(reply.readAll())
+with contextlib.suppress(json.JSONDecodeError):
+data = json.loads(buffer)
+return data
+return None
 def heartbeat(self):
 """
 Public method to check, if the 'ollama' server has started and is responsive.

Mercurial Repositories > ollama Interface / file comparison

comparison: OllamaInterface/OllamaClient.py

OllamaInterface/OllamaClient.py