--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/eric7/Debugger/DebuggerInterfacePython.py Sun Jul 24 11:29:56 2022 +0200
@@ -0,0 +1,1654 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2009 - 2022 Detlev Offenbach <detlev@die-offenbachs.de>
+#
+
+"""
+Module implementing the Python3 debugger interface for the debug server.
+"""
+
+import sys
+import os
+import logging
+import shlex
+import contextlib
+
+from PyQt6.QtCore import QObject, QProcess, QProcessEnvironment, QTimer
+
+from EricWidgets.EricApplication import ericApp
+from EricWidgets import EricMessageBox
+
+from . import DebugClientCapabilities
+
+import Globals
+import Preferences
+import Utilities
+
+from eric7config import getConfig
+
+
+ClientDefaultCapabilities = DebugClientCapabilities.HasAll
+
+
+class DebuggerInterfacePython(QObject):
+ """
+ Class implementing the debugger interface for the debug server for
+ Python 3.
+ """
+
+ def __init__(self, debugServer, passive):
+ """
+ Constructor
+
+ @param debugServer reference to the debug server
+ @type DebugServer
+ @param passive flag indicating passive connection mode
+ @type bool
+ """
+ super().__init__()
+
+ self.__isNetworked = True
+ self.__autoContinue = False
+ self.__autoContinued = []
+ self.__isStepCommand = False
+
+ self.debugServer = debugServer
+ self.passive = passive
+ self.process = None
+ self.__startedVenv = ""
+
+ self.queue = []
+ self.__master = None
+ self.__connections = {}
+ self.__pendingConnections = []
+ self.__inShutdown = False
+
+ # set default values for capabilities of clients
+ self.clientCapabilities = ClientDefaultCapabilities
+
+ # set translation function
+ self.translate = self.__identityTranslation
+
+ if passive:
+ # set translation function
+ if Preferences.getDebugger("PathTranslation"):
+ self.translateRemote = Preferences.getDebugger("PathTranslationRemote")
+ self.translateRemoteWindows = "\\" in self.translateRemote
+ self.translateLocal = Preferences.getDebugger("PathTranslationLocal")
+ self.translateLocalWindows = "\\" in self.translateLocal
+ self.translate = self.__remoteTranslation
+ else:
+ self.translate = self.__identityTranslation
+
+ # attribute to remember the name of the executed script
+ self.__scriptName = ""
+
+ def __identityTranslation(self, fn, remote2local=True):
+ """
+ Private method to perform the identity path translation.
+
+ @param fn filename to be translated
+ @type str
+ @param remote2local flag indicating the direction of translation
+ (False = local to remote, True = remote to local [default])
+ @type bool
+ @return translated filename
+ @rtype str
+ """
+ return fn
+
+ def __remoteTranslation(self, fn, remote2local=True):
+ """
+ Private method to perform the path translation.
+
+ @param fn filename to be translated
+ @type str
+ @param remote2local flag indicating the direction of translation
+ (False = local to remote, True = remote to local [default])
+ @type bool
+ @return translated filename
+ @rtype str
+ """
+ if remote2local:
+ path = fn.replace(self.translateRemote, self.translateLocal)
+ if self.translateLocalWindows:
+ path = path.replace("/", "\\")
+ else:
+ path = fn.replace(self.translateLocal, self.translateRemote)
+ if not self.translateRemoteWindows:
+ path = path.replace("\\", "/")
+
+ return path
+
+ def __startProcess(self, program, arguments, environment=None, workingDir=None):
+ """
+ Private method to start the debugger client process.
+
+ @param program name of the executable to start
+ @type str
+ @param arguments arguments to be passed to the program
+ @type list of str
+ @param environment dictionary of environment settings to pass
+ @type dict of str
+ @param workingDir directory to start the debugger client in
+ @type str
+ @return the process object
+ @rtype QProcess or None
+ """
+ proc = QProcess(self)
+ if environment is not None:
+ env = QProcessEnvironment()
+ for key, value in list(environment.items()):
+ env.insert(key, value)
+ proc.setProcessEnvironment(env)
+ args = arguments[:]
+ if workingDir:
+ proc.setWorkingDirectory(workingDir)
+ proc.start(program, args)
+ if not proc.waitForStarted(10000):
+ proc = None
+
+ return proc
+
+ def startRemote(
+ self,
+ port,
+ runInConsole,
+ venvName,
+ originalPathString,
+ workingDir=None,
+ configOverride=None,
+ ):
+ """
+ Public method to start a remote Python interpreter.
+
+ @param port port number the debug server is listening on
+ @type int
+ @param runInConsole flag indicating to start the debugger in a
+ console window
+ @type bool
+ @param venvName name of the virtual environment to be used
+ @type str
+ @param originalPathString original PATH environment variable
+ @type str
+ @param workingDir directory to start the debugger client in
+ @type str
+ @param configOverride dictionary containing the global config override
+ data
+ @type dict
+ @return client process object, a flag to indicate a network connection
+ and the name of the interpreter in case of a local execution
+ @rtype tuple of (QProcess, bool, str)
+ """
+ global origPathEnv
+
+ if not venvName:
+ venvName = Preferences.getDebugger("Python3VirtualEnv")
+ venvManager = ericApp().getObject("VirtualEnvManager")
+ interpreter = venvManager.getVirtualenvInterpreter(venvName)
+ execPath = venvManager.getVirtualenvExecPath(venvName)
+ if interpreter == "":
+ # use the interpreter used to run eric for identical variants
+ interpreter = Globals.getPythonExecutable()
+ if interpreter == "":
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr("""<p>No suitable Python3 environment configured.</p>"""),
+ )
+ return None, False, ""
+
+ self.__inShutdown = False
+
+ debugClientType = Preferences.getDebugger("DebugClientType3")
+ if debugClientType == "standard":
+ debugClient = os.path.join(
+ getConfig("ericDir"), "DebugClients", "Python", "DebugClient.py"
+ )
+ else:
+ debugClient = Preferences.getDebugger("DebugClient3")
+ if debugClient == "":
+ debugClient = os.path.join(
+ sys.path[0], "DebugClients", "Python", "DebugClient.py"
+ )
+
+ redirect = (
+ str(configOverride["redirect"])
+ if configOverride and configOverride["enable"]
+ else str(Preferences.getDebugger("Python3Redirect"))
+ )
+ noencoding = (
+ Preferences.getDebugger("Python3NoEncoding") and "--no-encoding" or ""
+ )
+ multiprocessEnabled = (
+ "--multiprocess" if Preferences.getDebugger("MultiProcessEnabled") else ""
+ )
+
+ if Preferences.getDebugger("RemoteDbgEnabled"):
+ ipaddr = self.debugServer.getHostAddress(False)
+ rexec = Preferences.getDebugger("RemoteExecution")
+ rhost = Preferences.getDebugger("RemoteHost")
+ if rhost == "":
+ rhost = "localhost"
+ if rexec:
+ args = Utilities.parseOptionString(rexec) + [
+ rhost,
+ interpreter,
+ debugClient,
+ ]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), redirect, ipaddr])
+ if Utilities.isWindowsPlatform():
+ if not os.path.splitext(args[0])[1]:
+ for ext in [".exe", ".com", ".cmd", ".bat"]:
+ prog = Utilities.getExecutablePath(args[0] + ext)
+ if prog:
+ args[0] = prog
+ break
+ else:
+ args[0] = Utilities.getExecutablePath(args[0])
+ process = self.__startProcess(args[0], args[1:], workingDir=workingDir)
+ if process is None:
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr(
+ """<p>The debugger backend could not be"""
+ """ started.</p>"""
+ ),
+ )
+
+ # set translation function
+ if Preferences.getDebugger("PathTranslation"):
+ self.translateRemote = Preferences.getDebugger(
+ "PathTranslationRemote"
+ )
+ self.translateRemoteWindows = "\\" in self.translateRemote
+ self.translateLocal = Preferences.getDebugger(
+ "PathTranslationLocal"
+ )
+ self.translate = self.__remoteTranslation
+ self.translateLocalWindows = "\\" in self.translateLocal
+ else:
+ self.translate = self.__identityTranslation
+ return process, self.__isNetworked, ""
+
+ # set translation function
+ self.translate = self.__identityTranslation
+
+ # setup the environment for the debugger
+ if Preferences.getDebugger("DebugEnvironmentReplace"):
+ clientEnv = {}
+ else:
+ clientEnv = os.environ.copy()
+ if originalPathString:
+ clientEnv["PATH"] = originalPathString
+ envlist = shlex.split(Preferences.getDebugger("DebugEnvironment"))
+ for el in envlist:
+ with contextlib.suppress(ValueError):
+ key, value = el.split("=", 1)
+ clientEnv[str(key)] = str(value)
+ if execPath:
+ if "PATH" in clientEnv:
+ clientEnv["PATH"] = os.pathsep.join([execPath, clientEnv["PATH"]])
+ else:
+ clientEnv["PATH"] = execPath
+
+ ipaddr = self.debugServer.getHostAddress(True)
+ if runInConsole or Preferences.getDebugger("ConsoleDbgEnabled"):
+ ccmd = Preferences.getDebugger("ConsoleDbgCommand")
+ if ccmd:
+ args = Utilities.parseOptionString(ccmd) + [
+ interpreter,
+ os.path.abspath(debugClient),
+ ]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), "0", ipaddr])
+ args[0] = Utilities.getExecutablePath(args[0])
+ process = self.__startProcess(
+ args[0], args[1:], clientEnv, workingDir=workingDir
+ )
+ if process is None:
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr(
+ """<p>The debugger backend could not be"""
+ """ started.</p>"""
+ ),
+ )
+ return process, self.__isNetworked, interpreter
+
+ args = [debugClient]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), redirect, ipaddr])
+ process = self.__startProcess(
+ interpreter, args, clientEnv, workingDir=workingDir
+ )
+ if process is None:
+ self.__startedVenv = ""
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr("""<p>The debugger backend could not be started.</p>"""),
+ )
+ else:
+ self.__startedVenv = venvName
+
+ return process, self.__isNetworked, interpreter
+
+ def startRemoteForProject(
+ self,
+ port,
+ runInConsole,
+ venvName,
+ originalPathString,
+ workingDir=None,
+ configOverride=None,
+ ):
+ """
+ Public method to start a remote Python interpreter for a project.
+
+ @param port port number the debug server is listening on
+ @type int
+ @param runInConsole flag indicating to start the debugger in a
+ console window
+ @type bool
+ @param venvName name of the virtual environment to be used
+ @type str
+ @param originalPathString original PATH environment variable
+ @type str
+ @param workingDir directory to start the debugger client in
+ @type str
+ @param configOverride dictionary containing the global config override
+ data
+ @type dict
+ @return client process object, a flag to indicate a network connection
+ and the name of the interpreter in case of a local execution
+ @rtype tuple of (QProcess, bool, str)
+ """
+ global origPathEnv
+
+ project = ericApp().getObject("Project")
+ if not project.isDebugPropertiesLoaded():
+ return None, self.__isNetworked, ""
+
+ # start debugger with project specific settings
+ debugClient = project.getDebugProperty("DEBUGCLIENT")
+
+ redirect = (
+ str(configOverride["redirect"])
+ if configOverride and configOverride["enable"]
+ else str(project.getDebugProperty("REDIRECT"))
+ )
+ noencoding = "--no-encoding" if project.getDebugProperty("NOENCODING") else ""
+ multiprocessEnabled = (
+ "--multiprocess" if Preferences.getDebugger("MultiProcessEnabled") else ""
+ )
+
+ if venvName:
+ venvManager = ericApp().getObject("VirtualEnvManager")
+ interpreter = venvManager.getVirtualenvInterpreter(venvName)
+ execPath = venvManager.getVirtualenvExecPath(venvName)
+ else:
+ venvName = project.getProjectVenv()
+ execPath = project.getProjectExecPath()
+ interpreter = project.getProjectInterpreter()
+ if interpreter == "":
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr("""<p>No suitable Python3 environment configured.</p>"""),
+ )
+ return None, self.__isNetworked, ""
+
+ self.__inShutdown = False
+
+ if project.getDebugProperty("REMOTEDEBUGGER"):
+ ipaddr = self.debugServer.getHostAddress(False)
+ rexec = project.getDebugProperty("REMOTECOMMAND")
+ rhost = project.getDebugProperty("REMOTEHOST")
+ if rhost == "":
+ rhost = "localhost"
+ if rexec:
+ args = Utilities.parseOptionString(rexec) + [
+ rhost,
+ interpreter,
+ debugClient,
+ ]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), redirect, ipaddr])
+ if Utilities.isWindowsPlatform():
+ if not os.path.splitext(args[0])[1]:
+ for ext in [".exe", ".com", ".cmd", ".bat"]:
+ prog = Utilities.getExecutablePath(args[0] + ext)
+ if prog:
+ args[0] = prog
+ break
+ else:
+ args[0] = Utilities.getExecutablePath(args[0])
+ process = self.__startProcess(args[0], args[1:], workingDir=workingDir)
+ if process is None:
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr(
+ """<p>The debugger backend could not be"""
+ """ started.</p>"""
+ ),
+ )
+ # set translation function
+ if project.getDebugProperty("PATHTRANSLATION"):
+ self.translateRemote = project.getDebugProperty("REMOTEPATH")
+ self.translateRemoteWindows = "\\" in self.translateRemote
+ self.translateLocal = project.getDebugProperty("LOCALPATH")
+ self.translateLocalWindows = "\\" in self.translateLocal
+ self.translate = self.__remoteTranslation
+ else:
+ self.translate = self.__identityTranslation
+ return process, self.__isNetworked, ""
+ else:
+ # remote shell command is missing
+ return None, self.__isNetworked, ""
+
+ # set translation function
+ self.translate = self.__identityTranslation
+
+ # setup the environment for the debugger
+ if project.getDebugProperty("ENVIRONMENTOVERRIDE"):
+ clientEnv = {}
+ else:
+ clientEnv = os.environ.copy()
+ if originalPathString:
+ clientEnv["PATH"] = originalPathString
+ envlist = shlex.split(project.getDebugProperty("ENVIRONMENTSTRING"))
+ for el in envlist:
+ with contextlib.suppress(ValueError):
+ key, value = el.split("=", 1)
+ clientEnv[str(key)] = str(value)
+ if execPath:
+ if "PATH" in clientEnv:
+ clientEnv["PATH"] = os.pathsep.join([execPath, clientEnv["PATH"]])
+ else:
+ clientEnv["PATH"] = execPath
+
+ ipaddr = self.debugServer.getHostAddress(True)
+ if runInConsole or project.getDebugProperty("CONSOLEDEBUGGER"):
+ ccmd = project.getDebugProperty(
+ "CONSOLECOMMAND"
+ ) or Preferences.getDebugger("ConsoleDbgCommand")
+ if ccmd:
+ args = Utilities.parseOptionString(ccmd) + [
+ interpreter,
+ os.path.abspath(debugClient),
+ ]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), "0", ipaddr])
+ args[0] = Utilities.getExecutablePath(args[0])
+ process = self.__startProcess(
+ args[0], args[1:], clientEnv, workingDir=workingDir
+ )
+ if process is None:
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr(
+ """<p>The debugger backend could not be"""
+ """ started.</p>"""
+ ),
+ )
+ return process, self.__isNetworked, interpreter
+
+ args = [debugClient]
+ if noencoding:
+ args.append(noencoding)
+ if multiprocessEnabled:
+ args.append(multiprocessEnabled)
+ args.extend([str(port), redirect, ipaddr])
+ process = self.__startProcess(
+ interpreter, args, clientEnv, workingDir=workingDir
+ )
+ if process is None:
+ self.__startedVenv = ""
+ EricMessageBox.critical(
+ None,
+ self.tr("Start Debugger"),
+ self.tr("""<p>The debugger backend could not be started.</p>"""),
+ )
+ else:
+ self.__startedVenv = venvName
+
+ return process, self.__isNetworked, interpreter
+
+ def getClientCapabilities(self):
+ """
+ Public method to retrieve the debug clients capabilities.
+
+ @return debug client capabilities
+ @rtype int
+ """
+ return self.clientCapabilities
+
+ def newConnection(self, sock):
+ """
+ Public slot to handle a new connection.
+
+ @param sock reference to the socket object
+ @type QTcpSocket
+ @return flag indicating success
+ @rtype bool
+ """
+ self.__pendingConnections.append(sock)
+
+ sock.readyRead.connect(lambda: self.__parseClientLine(sock))
+ sock.disconnected.connect(lambda: self.__socketDisconnected(sock))
+
+ return True
+
+ def __assignDebuggerId(self, sock, debuggerId):
+ """
+ Private method to set the debugger id for a recent debugger connection
+ attempt.
+
+ @param sock reference to the socket object
+ @type QTcpSocket
+ @param debuggerId id of the connected debug client
+ @type str
+ """
+ if sock in self.__pendingConnections:
+ self.__connections[debuggerId] = sock
+ self.__pendingConnections.remove(sock)
+
+ if self.__master is None:
+ self.__master = debuggerId
+ # Get the remote clients capabilities
+ self.remoteCapabilities(debuggerId)
+
+ self.debugServer.signalClientDebuggerId(debuggerId)
+
+ if debuggerId == self.__master:
+ self.__flush()
+ self.debugServer.masterClientConnected()
+
+ self.debugServer.initializeClient(debuggerId)
+
+ # perform auto-continue except for master
+ if (
+ debuggerId != self.__master
+ and self.__autoContinue
+ and not self.__isStepCommand
+ ):
+ self.__autoContinued.append(debuggerId)
+ QTimer.singleShot(0, lambda: self.remoteContinue(debuggerId))
+
+ def __socketDisconnected(self, sock):
+ """
+ Private slot handling a socket disconnecting.
+
+ @param sock reference to the disconnected socket
+ @type QTcpSocket
+ """
+ for debuggerId in self.__connections:
+ if self.__connections[debuggerId] is sock:
+ del self.__connections[debuggerId]
+ if debuggerId == self.__master:
+ self.__master = None
+ if debuggerId in self.__autoContinued:
+ self.__autoContinued.remove(debuggerId)
+ if not self.__inShutdown:
+ with contextlib.suppress(RuntimeError):
+ # can be ignored during a shutdown
+ self.debugServer.signalClientDisconnected(debuggerId)
+ break
+ else:
+ if sock in self.__pendingConnections:
+ self.__pendingConnections.remove(sock)
+
+ if not self.__connections:
+ # no active connections anymore
+ with contextlib.suppress(RuntimeError):
+ self.debugServer.signalLastClientExited()
+ # debug server object might have been deleted already
+ # ignore this
+ self.__autoContinued.clear()
+ self.debugServer.startClient()
+
+ def getDebuggerIds(self):
+ """
+ Public method to return the IDs of the connected debugger backends.
+
+ @return list of connected debugger backend IDs
+ @rtype list of str
+ """
+ return sorted(self.__connections.keys())
+
+ def __flush(self):
+ """
+ Private slot to flush the queue.
+ """
+ if self.__master:
+ # Send commands that were waiting for the connection.
+ for cmd in self.queue:
+ self.__writeJsonCommandToSocket(cmd, self.__connections[self.__master])
+
+ self.queue = []
+
+ def shutdown(self):
+ """
+ Public method to cleanly shut down.
+
+ It closes our sockets and shuts down the debug clients.
+ (Needed on Win OS)
+ """
+ if not self.__master:
+ return
+
+ self.__inShutdown = True
+
+ while self.__connections:
+ debuggerId, sock = self.__connections.popitem()
+ self.__shutdownSocket(sock)
+
+ while self.__pendingConnections:
+ sock = self.__pendingConnections.pop()
+ self.__shutdownSocket(sock)
+
+ # reinitialize
+ self.queue = []
+
+ self.__master = None
+
+ def __shutdownSocket(self, sock):
+ """
+ Private slot to shut down a socket.
+
+ @param sock reference to the socket
+ @type QTcpSocket
+ """
+ # do not want any slots called during shutdown
+ sock.readyRead.disconnect()
+ sock.disconnected.disconnect()
+
+ # close down socket, and shut down client as well.
+ self.__sendJsonCommand("RequestShutdown", {}, sock=sock)
+ sock.flush()
+ sock.close()
+
+ sock.setParent(None)
+ sock.deleteLater()
+ del sock
+
+ def isConnected(self):
+ """
+ Public method to test, if a debug client has connected.
+
+ @return flag indicating the connection status
+ @rtype bool
+ """
+ return bool(self.__connections)
+
+ def remoteEnvironment(self, env):
+ """
+ Public method to set the environment for a program to debug, run, ...
+
+ @param env environment settings
+ @type dict
+ """
+ self.__sendJsonCommand(
+ "RequestEnvironment", {"environment": env}, self.__master
+ )
+
+ def remoteLoad(
+ self,
+ fn,
+ argv,
+ wd,
+ traceInterpreter=False,
+ autoContinue=True,
+ enableMultiprocess=False,
+ ):
+ """
+ Public method to load a new program to debug.
+
+ @param fn the filename to debug
+ @type str
+ @param argv the commandline arguments to pass to the program
+ @type str
+ @param wd the working directory for the program
+ @type str
+ @param traceInterpreter flag indicating if the interpreter library
+ should be traced as well
+ @type bool
+ @param autoContinue flag indicating, that the debugger should not
+ stop at the first executable line
+ @type bool
+ @param enableMultiprocess flag indicating to perform multiprocess
+ debugging
+ @type bool
+ """
+ self.__autoContinue = autoContinue
+ self.__scriptName = os.path.abspath(fn)
+ self.__isStepCommand = False
+
+ wd = self.translate(wd, False)
+ fn = self.translate(os.path.abspath(fn), False)
+ self.__sendJsonCommand(
+ "RequestLoad",
+ {
+ "workdir": wd,
+ "filename": fn,
+ "argv": Utilities.parseOptionString(argv),
+ "traceInterpreter": traceInterpreter,
+ "multiprocess": enableMultiprocess,
+ },
+ self.__master,
+ )
+
+ def remoteRun(self, fn, argv, wd):
+ """
+ Public method to load a new program to run.
+
+ @param fn the filename to run
+ @type str
+ @param argv the commandline arguments to pass to the program
+ @type str
+ @param wd the working directory for the program
+ @type str
+ """
+ self.__scriptName = os.path.abspath(fn)
+
+ wd = self.translate(wd, False)
+ fn = self.translate(os.path.abspath(fn), False)
+ self.__sendJsonCommand(
+ "RequestRun",
+ {
+ "workdir": wd,
+ "filename": fn,
+ "argv": Utilities.parseOptionString(argv),
+ },
+ self.__master,
+ )
+
+ def remoteCoverage(self, fn, argv, wd, erase=False):
+ """
+ Public method to load a new program to collect coverage data.
+
+ @param fn the filename to run
+ @type str
+ @param argv the commandline arguments to pass to the program
+ @type str
+ @param wd the working directory for the program
+ @type str
+ @param erase flag indicating that coverage info should be
+ cleared first
+ @type bool
+ """
+ self.__scriptName = os.path.abspath(fn)
+
+ wd = self.translate(wd, False)
+ fn = self.translate(os.path.abspath(fn), False)
+ self.__sendJsonCommand(
+ "RequestCoverage",
+ {
+ "workdir": wd,
+ "filename": fn,
+ "argv": Utilities.parseOptionString(argv),
+ "erase": erase,
+ },
+ self.__master,
+ )
+
+ def remoteProfile(self, fn, argv, wd, erase=False):
+ """
+ Public method to load a new program to collect profiling data.
+
+ @param fn the filename to run
+ @type str
+ @param argv the commandline arguments to pass to the program
+ @type str
+ @param wd the working directory for the program
+ @type str
+ @param erase flag indicating that timing info should be cleared
+ first
+ @type bool
+ """
+ self.__scriptName = os.path.abspath(fn)
+
+ wd = self.translate(wd, False)
+ fn = self.translate(os.path.abspath(fn), False)
+ self.__sendJsonCommand(
+ "RequestProfile",
+ {
+ "workdir": wd,
+ "filename": fn,
+ "argv": Utilities.parseOptionString(argv),
+ "erase": erase,
+ },
+ self.__master,
+ )
+
+ def remoteStatement(self, debuggerId, stmt):
+ """
+ Public method to execute a Python statement.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param stmt the Python statement to execute.
+ @type str
+ """
+ self.__sendJsonCommand(
+ "ExecuteStatement",
+ {
+ "statement": stmt,
+ },
+ debuggerId,
+ )
+
+ def remoteStep(self, debuggerId):
+ """
+ Public method to single step the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__isStepCommand = True
+ self.__sendJsonCommand("RequestStep", {}, debuggerId)
+
+ def remoteStepOver(self, debuggerId):
+ """
+ Public method to step over the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__isStepCommand = True
+ self.__sendJsonCommand("RequestStepOver", {}, debuggerId)
+
+ def remoteStepOut(self, debuggerId):
+ """
+ Public method to step out the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__isStepCommand = True
+ self.__sendJsonCommand("RequestStepOut", {}, debuggerId)
+
+ def remoteStepQuit(self, debuggerId):
+ """
+ Public method to stop the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__isStepCommand = True
+ self.__sendJsonCommand("RequestStepQuit", {}, debuggerId)
+
+ def remoteContinue(self, debuggerId, special=False):
+ """
+ Public method to continue the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param special flag indicating a special continue operation
+ @type bool
+ """
+ self.__isStepCommand = False
+ self.__sendJsonCommand(
+ "RequestContinue",
+ {
+ "special": special,
+ },
+ debuggerId,
+ )
+
+ def remoteContinueUntil(self, debuggerId, line):
+ """
+ Public method to continue the debugged program to the given line
+ or until returning from the current frame.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param line the new line, where execution should be continued to
+ @type int
+ """
+ self.__isStepCommand = False
+ self.__sendJsonCommand(
+ "RequestContinueUntil",
+ {
+ "newLine": line,
+ },
+ debuggerId,
+ )
+
+ def remoteMoveIP(self, debuggerId, line):
+ """
+ Public method to move the instruction pointer to a different line.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param line the new line, where execution should be continued
+ @type int
+ """
+ self.__sendJsonCommand(
+ "RequestMoveIP",
+ {
+ "newLine": line,
+ },
+ debuggerId,
+ )
+
+ def remoteBreakpoint(
+ self, debuggerId, fn, line, setBreakpoint, cond=None, temp=False
+ ):
+ """
+ Public method to set or clear a breakpoint.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param fn filename the breakpoint belongs to
+ @type str
+ @param line linenumber of the breakpoint
+ @type int
+ @param setBreakpoint flag indicating setting or resetting a breakpoint
+ @type bool
+ @param cond condition of the breakpoint
+ @type str
+ @param temp flag indicating a temporary breakpoint
+ @type bool
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ self.__sendJsonCommand(
+ "RequestBreakpoint",
+ {
+ "filename": self.translate(fn, False),
+ "line": line,
+ "temporary": temp,
+ "setBreakpoint": setBreakpoint,
+ "condition": cond,
+ },
+ debuggerId,
+ )
+
+ def remoteBreakpointEnable(self, debuggerId, fn, line, enable):
+ """
+ Public method to enable or disable a breakpoint.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param fn filename the breakpoint belongs to
+ @type str
+ @param line linenumber of the breakpoint
+ @type int
+ @param enable flag indicating enabling or disabling a breakpoint
+ @type bool
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ self.__sendJsonCommand(
+ "RequestBreakpointEnable",
+ {
+ "filename": self.translate(fn, False),
+ "line": line,
+ "enable": enable,
+ },
+ debuggerId,
+ )
+
+ def remoteBreakpointIgnore(self, debuggerId, fn, line, count):
+ """
+ Public method to ignore a breakpoint the next couple of occurrences.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param fn filename the breakpoint belongs to
+ @type str
+ @param line linenumber of the breakpoint
+ @type int
+ @param count number of occurrences to ignore
+ @type int
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ self.__sendJsonCommand(
+ "RequestBreakpointIgnore",
+ {
+ "filename": self.translate(fn, False),
+ "line": line,
+ "count": count,
+ },
+ debuggerId,
+ )
+
+ def remoteWatchpoint(self, debuggerId, cond, setWatch, temp=False):
+ """
+ Public method to set or clear a watch expression.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param cond expression of the watch expression
+ @type str
+ @param setWatch flag indicating setting or resetting a watch expression
+ @type bool
+ @param temp flag indicating a temporary watch expression
+ @type bool
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ # cond is combination of cond and special (s. watch expression
+ # viewer)
+ self.__sendJsonCommand(
+ "RequestWatch",
+ {
+ "temporary": temp,
+ "setWatch": setWatch,
+ "condition": cond,
+ },
+ debuggerId,
+ )
+
+ def remoteWatchpointEnable(self, debuggerId, cond, enable):
+ """
+ Public method to enable or disable a watch expression.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param cond expression of the watch expression
+ @type str
+ @param enable flag indicating enabling or disabling a watch expression
+ @type bool
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ # cond is combination of cond and special (s. watch expression
+ # viewer)
+ self.__sendJsonCommand(
+ "RequestWatchEnable",
+ {
+ "condition": cond,
+ "enable": enable,
+ },
+ debuggerId,
+ )
+
+ def remoteWatchpointIgnore(self, debuggerId, cond, count):
+ """
+ Public method to ignore a watch expression the next couple of
+ occurrences.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param cond expression of the watch expression
+ @type str
+ @param count number of occurrences to ignore
+ @type int
+ """
+ debuggerList = [debuggerId] if debuggerId else list(self.__connections.keys())
+ for debuggerId in debuggerList:
+ # cond is combination of cond and special (s. watch expression
+ # viewer)
+ self.__sendJsonCommand(
+ "RequestWatchIgnore",
+ {
+ "condition": cond,
+ "count": count,
+ },
+ debuggerId,
+ )
+
+ def remoteRawInput(self, debuggerId, inputString):
+ """
+ Public method to send the raw input to the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param inputString the raw input
+ @type str
+ """
+ self.__sendJsonCommand(
+ "RawInput",
+ {
+ "input": inputString,
+ },
+ debuggerId,
+ )
+
+ def remoteThreadList(self, debuggerId):
+ """
+ Public method to request the list of threads from the client.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__sendJsonCommand("RequestThreadList", {}, debuggerId)
+
+ def remoteSetThread(self, debuggerId, tid):
+ """
+ Public method to request to set the given thread as current thread.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param tid id of the thread
+ @type int
+ """
+ self.__sendJsonCommand(
+ "RequestThreadSet",
+ {
+ "threadID": tid,
+ },
+ debuggerId,
+ )
+
+ def remoteClientStack(self, debuggerId):
+ """
+ Public method to request the stack of the main thread.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__sendJsonCommand("RequestStack", {}, debuggerId)
+
+ def remoteClientVariables(
+ self, debuggerId, scope, filterList, framenr=0, maxSize=0
+ ):
+ """
+ Public method to request the variables of the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param scope the scope of the variables (0 = local, 1 = global)
+ @type int
+ @param filterList list of variable types to filter out
+ @type list of str
+ @param framenr framenumber of the variables to retrieve
+ @type int
+ @param maxSize maximum size the formatted value of a variable will
+ be shown. If it is bigger than that, a 'too big' indication will
+ be given (@@TOO_BIG_TO_SHOW@@).
+ @type int
+ """
+ self.__sendJsonCommand(
+ "RequestVariables",
+ {
+ "frameNumber": framenr,
+ "scope": scope,
+ "filters": filterList,
+ "maxSize": maxSize,
+ },
+ debuggerId,
+ )
+
+ def remoteClientVariable(
+ self, debuggerId, scope, filterList, var, framenr=0, maxSize=0
+ ):
+ """
+ Public method to request the variables of the debugged program.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param scope the scope of the variables (0 = local, 1 = global)
+ @type int
+ @param filterList list of variable types to filter out
+ @type list of str
+ @param var list encoded name of variable to retrieve
+ @type list of str
+ @param framenr framenumber of the variables to retrieve
+ @type int
+ @param maxSize maximum size the formatted value of a variable will
+ be shown. If it is bigger than that, a 'too big' indication will
+ be given (@@TOO_BIG_TO_SHOW@@).
+ @type int
+ """
+ self.__sendJsonCommand(
+ "RequestVariable",
+ {
+ "variable": var,
+ "frameNumber": framenr,
+ "scope": scope,
+ "filters": filterList,
+ "maxSize": maxSize,
+ },
+ debuggerId,
+ )
+
+ def remoteClientDisassembly(self, debuggerId):
+ """
+ Public method to ask the client for the latest traceback disassembly.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__sendJsonCommand("RequestDisassembly", {}, debuggerId)
+
+ def remoteClientSetFilter(self, debuggerId, scope, filterStr):
+ """
+ Public method to set a variables filter list.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param scope the scope of the variables (0 = local, 1 = global)
+ @type int
+ @param filterStr regexp string for variable names to filter out
+ @type str
+ """
+ self.__sendJsonCommand(
+ "RequestSetFilter",
+ {
+ "scope": scope,
+ "filter": filterStr,
+ },
+ debuggerId,
+ )
+
+ def setCallTraceEnabled(self, debuggerId, on):
+ """
+ Public method to set the call trace state.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param on flag indicating to enable the call trace function
+ @type bool
+ """
+ self.__sendJsonCommand(
+ "RequestCallTrace",
+ {
+ "enable": on,
+ },
+ debuggerId,
+ )
+
+ def remoteNoDebugList(self, debuggerId, noDebugList):
+ """
+ Public method to set a list of programs not to be debugged.
+
+ The programs given in the list will not be run under the control
+ of the multi process debugger.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param noDebugList list of Python programs not to be debugged
+ @type list of str
+ """
+ self.__sendJsonCommand(
+ "RequestSetNoDebugList",
+ {
+ "noDebug": noDebugList,
+ },
+ debuggerId,
+ )
+
+ def remoteBanner(self):
+ """
+ Public slot to get the banner info of the remote client.
+ """
+ self.__sendJsonCommand("RequestBanner", {})
+
+ def remoteCapabilities(self, debuggerId):
+ """
+ Public slot to get the debug clients capabilities.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ """
+ self.__sendJsonCommand("RequestCapabilities", {}, debuggerId)
+
+ def remoteCompletion(self, debuggerId, text):
+ """
+ Public slot to get the a list of possible commandline completions
+ from the remote client.
+
+ @param debuggerId ID of the debugger backend
+ @type str
+ @param text the text to be completed
+ @type str
+ """
+ self.__sendJsonCommand(
+ "RequestCompletion",
+ {
+ "text": text,
+ },
+ debuggerId,
+ )
+
+ def __parseClientLine(self, sock):
+ """
+ Private method to handle data from the client.
+
+ @param sock reference to the socket to read data from
+ @type QTcpSocket
+ """
+ while sock and sock.canReadLine():
+ qs = sock.readLine()
+ line = bytes(qs).decode(encoding=Preferences.getSystem("StringEncoding"))
+
+ logging.debug("<Debug-Server> %s", line)
+ ##print("Server: ", line) ## debug # __IGNORE_WARNING_M891__
+
+ self.__handleJsonCommand(line, sock)
+
+ def __handleJsonCommand(self, jsonStr, sock):
+ """
+ Private method to handle a command or response serialized as a
+ JSON string.
+
+ @param jsonStr string containing the command or response received
+ from the debug backend
+ @type str
+ @param sock reference to the socket the data was received from
+ @type QTcpSocket
+ """
+ import json
+
+ try:
+ commandDict = json.loads(jsonStr.strip())
+ except (TypeError, ValueError) as err:
+ EricMessageBox.critical(
+ None,
+ self.tr("Debug Protocol Error"),
+ self.tr(
+ """<p>The response received from the debugger"""
+ """ backend could not be decoded. Please report"""
+ """ this issue with the received data to the"""
+ """ eric bugs email address.</p>"""
+ """<p>Error: {0}</p>"""
+ """<p>Data:<br/>{1}</p>"""
+ ).format(str(err), Utilities.html_encode(jsonStr.strip())),
+ EricMessageBox.Ok,
+ )
+ return
+
+ method = commandDict["method"]
+ params = commandDict["params"]
+
+ if method == "DebuggerId":
+ self.__assignDebuggerId(sock, params["debuggerId"])
+
+ elif method == "ClientOutput":
+ self.debugServer.signalClientOutput(params["text"], params["debuggerId"])
+
+ elif method in ["ResponseLine", "ResponseStack"]:
+ # Check if obsolete thread was clicked
+ if params["stack"] == []:
+ # Request updated list
+ self.remoteThreadList(params["debuggerId"])
+ return
+ for s in params["stack"]:
+ s[0] = self.translate(s[0], True)
+ cf = params["stack"][0]
+ if self.__autoContinue and params["debuggerId"] not in self.__autoContinued:
+ self.__autoContinued.append(params["debuggerId"])
+ QTimer.singleShot(0, lambda: self.remoteContinue(params["debuggerId"]))
+ else:
+ self.debugServer.signalClientLine(
+ cf[0],
+ int(cf[1]),
+ params["debuggerId"],
+ method == "ResponseStack",
+ threadName=params["threadName"],
+ )
+ self.debugServer.signalClientStack(
+ params["stack"],
+ params["debuggerId"],
+ threadName=params["threadName"],
+ )
+
+ elif method == "CallTrace":
+ isCall = params["event"].lower() == "c"
+ fromInfo = params["from"]
+ toInfo = params["to"]
+ self.debugServer.signalClientCallTrace(
+ isCall,
+ fromInfo["filename"],
+ str(fromInfo["linenumber"]),
+ fromInfo["codename"],
+ toInfo["filename"],
+ str(toInfo["linenumber"]),
+ toInfo["codename"],
+ params["debuggerId"],
+ )
+
+ elif method == "ResponseVariables":
+ self.debugServer.signalClientVariables(
+ params["scope"], params["variables"], params["debuggerId"]
+ )
+
+ elif method == "ResponseVariable":
+ self.debugServer.signalClientVariable(
+ params["scope"],
+ [params["variable"]] + params["variables"],
+ params["debuggerId"],
+ )
+
+ elif method == "ResponseThreadList":
+ self.debugServer.signalClientThreadList(
+ params["currentID"], params["threadList"], params["debuggerId"]
+ )
+
+ elif method == "ResponseThreadSet":
+ self.debugServer.signalClientThreadSet(params["debuggerId"])
+
+ elif method == "ResponseCapabilities":
+ self.clientCapabilities = params["capabilities"]
+ if params["debuggerId"] == self.__master:
+ # signal only for the master connection
+ self.debugServer.signalClientCapabilities(
+ params["capabilities"],
+ params["clientType"],
+ self.__startedVenv,
+ )
+
+ elif method == "ResponseBanner":
+ if params["debuggerId"] == self.__master:
+ # signal only for the master connection
+ self.debugServer.signalClientBanner(
+ params["version"],
+ params["platform"],
+ self.__startedVenv,
+ )
+
+ elif method == "ResponseOK":
+ self.debugServer.signalClientStatement(False, params["debuggerId"])
+
+ elif method == "ResponseContinue":
+ self.debugServer.signalClientStatement(True, params["debuggerId"])
+
+ elif method == "RequestRaw":
+ self.debugServer.signalClientRawInput(
+ params["prompt"], params["echo"], params["debuggerId"]
+ )
+
+ elif method == "ResponseBPConditionError":
+ fn = self.translate(params["filename"], True)
+ self.debugServer.signalClientBreakConditionError(
+ fn, params["line"], params["debuggerId"]
+ )
+
+ elif method == "ResponseClearBreakpoint":
+ fn = self.translate(params["filename"], True)
+ self.debugServer.signalClientClearBreak(
+ fn, params["line"], params["debuggerId"]
+ )
+
+ elif method == "ResponseWatchConditionError":
+ self.debugServer.signalClientWatchConditionError(
+ params["condition"], params["debuggerId"]
+ )
+
+ elif method == "ResponseClearWatch":
+ self.debugServer.signalClientClearWatch(
+ params["condition"], params["debuggerId"]
+ )
+
+ elif method == "ResponseDisassembly":
+ self.debugServer.signalClientDisassembly(
+ params["disassembly"], params["debuggerId"]
+ )
+
+ elif method == "ResponseException":
+ exctype = params["type"]
+ excmessage = params["message"]
+ stack = params["stack"]
+ if stack:
+ for stackEntry in stack:
+ stackEntry[0] = self.translate(stackEntry[0], True)
+ if stack[0] and stack[0][0] == "<string>":
+ for stackEntry in stack:
+ if stackEntry[0] == "<string>":
+ stackEntry[0] = self.__scriptName
+ else:
+ break
+
+ self.debugServer.signalClientException(
+ exctype, excmessage, stack, params["debuggerId"], params["threadName"]
+ )
+
+ elif method == "ResponseSyntax":
+ self.debugServer.signalClientSyntaxError(
+ params["message"],
+ self.translate(params["filename"], True),
+ params["linenumber"],
+ params["characternumber"],
+ params["debuggerId"],
+ params["threadName"],
+ )
+
+ elif method == "ResponseSignal":
+ self.debugServer.signalClientSignal(
+ params["message"],
+ self.translate(params["filename"], True),
+ params["linenumber"],
+ params["function"],
+ params["arguments"],
+ params["debuggerId"],
+ )
+
+ elif method == "ResponseExit":
+ self.__scriptName = ""
+ self.debugServer.signalClientExit(
+ params["program"],
+ params["status"],
+ params["message"],
+ params["debuggerId"],
+ )
+ if params["debuggerId"] == self.__master:
+ self.debugServer.signalMainClientExit()
+
+ elif method == "PassiveStartup":
+ self.debugServer.passiveStartUp(
+ self.translate(params["filename"], True),
+ params["exceptions"],
+ params["debuggerId"],
+ )
+
+ elif method == "ResponseCompletion":
+ self.debugServer.signalClientCompletionList(
+ params["completions"], params["text"], params["debuggerId"]
+ )
+
+ def __sendJsonCommand(self, command, params, debuggerId="", sock=None):
+ """
+ Private method to send a single command to the client.
+
+ @param command command name to be sent
+ @type str
+ @param params dictionary of named parameters for the command
+ @type dict
+ @param debuggerId id of the debug client to send the command to
+ @type str
+ @param sock reference to the socket object to be used (only used if
+ debuggerId is not given)
+ @type QTcpSocket
+ """
+ import json
+
+ commandDict = {
+ "jsonrpc": "2.0",
+ "method": command,
+ "params": params,
+ }
+ cmd = json.dumps(commandDict) + "\n"
+
+ if debuggerId and debuggerId in self.__connections:
+ sock = self.__connections[debuggerId]
+ elif sock is None and self.__master is not None:
+ sock = self.__connections[self.__master]
+ if sock is not None:
+ self.__writeJsonCommandToSocket(cmd, sock)
+ else:
+ self.queue.append(cmd)
+
+ def __writeJsonCommandToSocket(self, cmd, sock):
+ """
+ Private method to write a JSON command to the socket.
+
+ @param cmd JSON command to be sent
+ @type str
+ @param sock reference to the socket to write to
+ @type QTcpSocket
+ """
+ data = cmd.encode("utf8", "backslashreplace")
+ length = "{0:09d}".format(len(data))
+ sock.write(length.encode() + data)
+ sock.flush()
+
+
+def createDebuggerInterfacePython3(debugServer, passive):
+ """
+ Module function to create a debugger interface instance.
+
+
+ @param debugServer reference to the debug server
+ @type DebugServer
+ @param passive flag indicating passive connection mode
+ @type bool
+ @return instantiated debugger interface
+ @rtype DebuggerInterfacePython
+ """
+ return DebuggerInterfacePython(debugServer, passive)
+
+
+def getRegistryData():
+ """
+ Module function to get characterizing data for the supported debugger
+ interfaces.
+
+ @return list of tuples containing the client type, the client capabilities,
+ the client file type associations and a reference to the creation
+ function
+ @rtype list of tuple of (str, int, list of str, function)
+ """
+ py3Exts = []
+ for ext in Preferences.getDebugger("Python3Extensions").split():
+ if ext.startswith("."):
+ py3Exts.append(ext)
+ else:
+ py3Exts.append(".{0}".format(ext))
+
+ registryData = []
+ if py3Exts:
+ registryData.append(
+ (
+ "Python3",
+ ClientDefaultCapabilities,
+ py3Exts,
+ createDebuggerInterfacePython3,
+ )
+ )
+
+ return registryData
