--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/eric7/RemoteServerInterface/EricServerFileSystemInterface.py Fri Feb 02 11:29:08 2024 +0100
@@ -0,0 +1,345 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
+#
+
+"""
+Module implementing the file system interface to the eric-ide server.
+"""
+
+import contextlib
+
+from PyQt6.QtCore import QEventLoop, QObject
+
+from eric7.RemoteServer.EricRequestCategory import EricRequestCategory
+
+
+class EricServerFileSystemInterface(QObject):
+ """
+ Class implementing the file system interface to the eric-ide server.
+ """
+
+ def __init__(self, serverInterface):
+ """
+ Constructor
+
+ @param serverInterface reference to the eric-ide server interface
+ @type EricServerInterface
+ """
+ super().__init__(parent=serverInterface)
+
+ self.__serverInterface = serverInterface
+
+ def getcwd(self):
+ """
+ Public method to get the current working directory of the eric-ide server.
+
+ @return current working directory of the eric-ide server
+ @rtype str
+ """
+ loop = QEventLoop()
+ cwd = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal cwd
+
+ if reply == "Getcwd":
+ cwd = params["directory"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Getcwd",
+ params={},
+ callback=callback,
+ )
+
+ loop.exec()
+ return cwd
+
+ def chdir(self, directory):
+ """
+ Public method to change the current working directory of the eric-ide server.
+
+ @param directory absolute path of the working directory to change to
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ loop = QEventLoop()
+ ok = False
+ error = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal ok, error
+
+ if reply == "Chdir":
+ ok = params["ok"]
+ with contextlib.suppress(KeyError):
+ error = params["error"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Chdir",
+ params={"directory": directory},
+ callback=callback,
+ )
+
+ loop.exec()
+ return ok, error
+
+ def listdir(self, directory=""):
+ """
+ Public method to get a directory listing.
+
+ @param directory directory to be listed. An empty directory means to list
+ the eric-ide server current directory. (defaults to "")
+ @type str (optional)
+ @return tuple containing the listed directory, the path separartor and the
+ directory listing. Each directory listing entry contains a dictionary
+ with the relevant data.
+ @rtype tuple of (str, str, dict)
+ """
+ if directory is None:
+ # sanitize the directory in case it is None
+ directory = ""
+
+ loop = QEventLoop()
+ listedDirectory = ""
+ separator = ""
+ listing = []
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal listedDirectory, listing, separator
+
+ if reply == "Listdir":
+ listedDirectory = params["directory"]
+ listing = params["listing"]
+ separator = params["separator"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Listdir",
+ params={"directory": directory},
+ callback=callback,
+ )
+
+ loop.exec()
+ return listedDirectory, separator, listing
+
+ def mkdir(self, directory):
+ """
+ Public method to create a new directory on the eric-ide server.
+
+ @param directory absolute path of the new directory
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ loop = QEventLoop()
+ ok = False
+ error = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal ok, error
+
+ if reply == "Mkdir":
+ ok = params["ok"]
+ with contextlib.suppress(KeyError):
+ error = params["error"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Mkdir",
+ params={"directory": directory},
+ callback=callback,
+ )
+
+ loop.exec()
+ return ok, error
+
+ def rmdir(self, directory):
+ """
+ Public method to delete a directory on the eric-ide server.
+
+ @param directory absolute path of the directory
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ loop = QEventLoop()
+ ok = False
+ error = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal ok, error
+
+ if reply == "Rmdir":
+ ok = params["ok"]
+ with contextlib.suppress(KeyError):
+ error = params["error"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Rmdir",
+ params={"directory": directory},
+ callback=callback,
+ )
+
+ loop.exec()
+ return ok, error
+
+ def replace(self, oldName, newName):
+ """
+ Public method to rename a file or directory.
+
+ @param oldName current name of the file or directory
+ @type str
+ @param newName new name for the file or directory
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ loop = QEventLoop()
+ ok = False
+ error = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal ok, error
+
+ if reply == "Replace":
+ ok = params["ok"]
+ with contextlib.suppress(KeyError):
+ error = params["error"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Replace",
+ params={"old_name": oldName, "new_name": newName},
+ callback=callback,
+ )
+
+ loop.exec()
+ return ok, error
+
+ def remove(self, filename):
+ """
+ Public method to delete a file on the eric-ide server.
+
+ @param filename absolute path of the file
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ loop = QEventLoop()
+ ok = False
+ error = ""
+
+ def callback(reply, params):
+ """
+ Function to handle the server reply
+
+ @param reply name of the server reply
+ @type str
+ @param params dictionary containing the reply data
+ @type dict
+ """
+ nonlocal ok, error
+
+ if reply == "Remove":
+ ok = params["ok"]
+ with contextlib.suppress(KeyError):
+ error = params["error"]
+ loop.quit()
+
+ self.__serverInterface.sendJson(
+ category=EricRequestCategory.FileSystem,
+ request="Remove",
+ params={"filename": filename},
+ callback=callback,
+ )
+
+ loop.exec()
+ return ok, error
+
+ #######################################################################
+ ## Methods for reading and writing files
+ #######################################################################
+
+ def readFile(self, filename):
+ """
+ Public method to read a file from the eric-ide server.
+
+ @param filename name of the file to read
+ @type str
+ @return tuple containing an OK flag, the read data and an error string in
+ case of an issue
+ @rtype tuple of (bool, str, str)
+ """
+ # TODO: 'readFile()' not implemented yet
+
+ def writeFile(self, filename, data):
+ """
+ Public method to write the data to a file on the eric-ide server.
+
+ @param filename name of the file to write
+ @type str
+ @param data data to be written
+ @type str
+ @return tuple containing an OK flag and an error string in case of an issue
+ @rtype tuple of (bool, str)
+ """
+ # TODO: 'writeFile()' not implemented yet
