eric: comparison DebugClients/Python3/DebugClientBase.py

-:1b59c4ba121e
+:8cd4d08fa9f6
 DebugClientInstance = None
 ################################################################################
-def DebugClientInput(prompt = "", echo = True):
+def DebugClientInput(prompt="", echo=True):
 """
 Replacement for the standard input builtin.
 This function works with the split debugger.
 DebugClientOrigInput = __main__.__builtins__.__dict__['input']
 __main__.__builtins__.__dict__['input'] = DebugClientInput
 ################################################################################
 def DebugClientFork():
 """
 Replacement for the standard os.fork().
 """
 if DebugClientInstance is None:
 DebugClientOrigFork = os.fork
 os.fork = DebugClientFork
 ################################################################################
 def DebugClientClose(fd):
 """
 Replacement for the standard os.close(fd).
 @param fd open file descriptor to be closed (integer)
 if 'close' in dir(os):
 DebugClientOrigClose = os.close
 os.close = DebugClientClose
 ################################################################################
 def DebugClientSetRecursionLimit(limit):
 """
 Replacement for the standard sys.setrecursionlimit(limit).
 sys.setrecursionlimit = DebugClientSetRecursionLimit
 DebugClientSetRecursionLimit(sys.getrecursionlimit())
 ################################################################################
 class DebugClientBase(object):
 """
 Class implementing the client side of the debugger.
 It provides access to the Python interpeter from a debugger running in another
 statements.  Commands and statement are always exactly one line and may be
 interspersed.
 The protocol is as follows.  First the client opens a connection to the
 debugger and then sends a series of one line commands.  A command is either
 &gt;Load&lt;, &gt;Step&lt;, &gt;StepInto&lt;, ... or a Python statement.
 See DebugProtocol.py for a listing of valid protocol tokens.
 A Python statement consists of the statement to execute, followed (in a
 separate line) by &gt;OK?&lt;.  If the statement was incomplete then the response
 is &gt;Continue&lt;.  If there was an exception then the response is
 &gt;Exception&lt;.
 Otherwise the response is &gt;OK&lt;.  The reason for the &gt;OK?&lt; part is to
 provide a sentinal (ie. the responding &gt;OK&lt;) after any possible output as a
 result of executing the command.
 if m:
 self.__coding = m.group(1)
 return
 self.__coding = default
-def attachThread(self, target = None, args = None, kwargs = None, mainThread = False):
+def attachThread(self, target=None, args=None, kwargs=None, mainThread=False):
 """
 Public method to setup a thread for DebugClient to debug.
 If mainThread is non-zero, then we are attaching to the already
 started mainthread of the app and the rest of the args are ignored.
 @param target the start function of the target thread (i.e. the user code)
 @param args arguments to pass to target
 @param kwargs keyword arguments to pass to target
 @param mainThread True, if we are attaching to the already
 started mainthread of the app
 @return The identifier of the created thread
 """
 if self.debugging:
 sys.setprofile(self.profile)
 d["broken"] = self.isBroken()
 threadList.append(d)
 self.write("{0}{1!r}\n".format(ResponseThreadList, (currentId, threadList)))
-def input(self, prompt, echo = True):
+def input(self, prompt, echo=True):
 """
 Public method to implement input() using the event loop.
 @param prompt the prompt to be shown (string)
 @param echo Flag indicating echoing of the input (boolean)
 It ensures that the debug server is informed of the raised exception.
 """
 self.pendingResponse = ResponseException
-def sessionClose(self, exit = True):
+def sessionClose(self, exit=True):
 """
 Public method to close the session with the debugger and optionally terminate.
 @param exit flag indicating to terminate (boolean)
 """
 if exit:
 # Ok, go away.
 sys.exit()
-def __compileFileSource(self, filename, mode = 'exec'):
+def __compileFileSource(self, filename, mode='exec'):
 """
 Private method to compile source code read from a file.
 @param filename name of the source file (string)
 @param mode kind of code to be generated (string, exec or eval)
 @return compiled code object (None in case of errors)
 """
-with open(filename, encoding = self.__coding) as fp:
+with open(filename, encoding=self.__coding) as fp:
 statement = fp.read()
 try:
 code = compile(statement + '\n', filename, mode)
 except SyntaxError:
 self.write("{0}{1}\n".format(ResponseSyntax, str(exclist)))
 return None
 return code
-def handleLine(self,line):
+def handleLine(self, line):
 """
 Public method to handle the receipt of a complete line.
 It first looks for a valid protocol token at the start of the line. Thereafter
 it trys to execute the lines accumulated so far.
 self.mainFrame = None
 self.inRawMode = False
 self.debugging = True
 self.threads.clear()
-self.attachThread(mainThread = True)
+self.attachThread(mainThread=True)
 # set the system exception handling function to ensure, that
 # we report on all unhandled exceptions
 sys.excepthook = self.__unhandled_exception
 self.mainThread.tracePython = tracePython
 # This will eventually enter a local event loop.
 # Note the use of backquotes to cause a repr of self.running. The
 # need for this is on Windows os where backslash is the path separator.
 # They will get inadvertantly stripped away during the eval causing
 # IOErrors, if self.running is passed as a normal str.
 self.debugMod.__dict__['__file__'] = self.running
 sys.modules['__main__'] = self.debugMod
 code = self.__compileFileSource(self.running)
 if code:
 self.mainFrame = None
 self.botframe = None
 self.inRawMode = False
 self.threads.clear()
-self.attachThread(mainThread = True)
+self.attachThread(mainThread=True)
 # set the system exception handling function to ensure, that
 # we report on all unhandled exceptions
 sys.excepthook = self.__unhandled_exception
 self.mainThread.tracePython = False
 self.debugMod.__dict__['__file__'] = sys.argv[0]
 sys.modules['__main__'] = self.debugMod
-exec(open(sys.argv[0], encoding = self.__coding).read(),
+exec(open(sys.argv[0], encoding=self.__coding).read(),
 self.debugMod.__dict__)
 self.writestream.flush()
 return
 if cmd == RequestProfile:
 if int(erase):
 self.prof.erase()
 self.debugMod.__dict__['__file__'] = sys.argv[0]
 sys.modules['__main__'] = self.debugMod
-fp = open(sys.argv[0], encoding = self.__coding)
+fp = open(sys.argv[0], encoding=self.__coding)
 try:
 script = fp.read()
 finally:
 fp.close()
 if script:
 # set the system exception handling function to ensure, that
 # we report on all unhandled exceptions
 sys.excepthook = self.__unhandled_exception
 # generate a coverage object
-self.cover = coverage(auto_data = True,
+self.cover = coverage(auto_data=True,
-data_file = "{0}.coverage".format(os.path.splitext(sys.argv[0])[0]))
+data_file="{0}.coverage".format(os.path.splitext(sys.argv[0])[0]))
 self.cover.use_cache(True)
 if int(erase):
 self.cover.erase()
 sys.modules['__main__'] = self.debugMod
 self.debugMod.__dict__['__file__'] = sys.argv[0]
 self.cover.start()
-exec(open(sys.argv[0], encoding = self.__coding).read(),
+exec(open(sys.argv[0], encoding=self.__coding).read(),
 self.debugMod.__dict__)
 self.cover.stop()
 self.cover.save()
 self.writestream.flush()
 return
 self.write(ResponseException + '\n')
 return
 if cmd == RequestBanner:
 self.write('{0}{1}\n'.format(ResponseBanner,
 str(("Python {0}".format(sys.version),
 socket.gethostname(), self.variant))))
 return
 if cmd == RequestCapabilities:
 self.write('{0}{1:d}, "Python3"\n'.format(ResponseCapabilities,
 self.__clientCapabilities()))
 return
 if cmd == RequestCompletion:
 self.__completionList(arg.replace("u'", "'"))
 return
 # generate a coverage object
 if int(cov):
 from coverage import coverage
-self.cover = coverage(auto_data = True,
+self.cover = coverage(auto_data=True,
-data_file = "{0}.coverage".format(os.path.splitext(covname)[0]))
+data_file="{0}.coverage".format(os.path.splitext(covname)[0]))
 self.cover.use_cache(True)
 if int(erase):
 self.cover.erase()
 else:
 self.cover = None
 self.__exceptionRaised()
 else:
 if code is None:
 self.pendingResponse = ResponseContinue
 else:
 self.buffer = ''
 try:
 if self.running is None:
 exec(code, self.debugMod.__dict__)
 pass
 return self.clientCapabilities
 except ImportError:
 return self.clientCapabilities & ~DebugClientCapabilities.HasProfiler
-def write(self,s):
+def write(self, s):
 """
 Public method to write data to the output stream.
 @param s data to be written (string)
 """
 if not self.passive:
 # At this point simulate an event loop.
 self.eventLoop()
-def eventLoop(self, disablePolling = False):
+def eventLoop(self, disablePolling=False):
 """
 Public method implementing our event loop.
 @param disablePolling flag indicating to enter an event loop with
 polling disabled (boolean)
 self.writeReady(self.writestream.fileno())
 if self.errorstream in wrdy:
 self.writeReady(self.errorstream.fileno())
-def connectDebugger(self, port, remoteAddress = None, redirect = True):
+def connectDebugger(self, port, remoteAddress=None, redirect=True):
 """
 Public method to establish a session with the debugger.
 It opens a network connection to the debugger, connects it to stdin,
 stdout and stderr and saves these file objects in case the application
 being debugged redirects them itself.
 @param port the port number to connect to (int)
 @param remoteAddress the network address of the debug server host (string)
 sys.stdout = self.writestream
 sys.stderr = self.errorstream
 self.redirect = redirect
 # attach to the main thread here
-self.attachThread(mainThread = True)
+self.attachThread(mainThread=True)
 def __unhandled_exception(self, exctype, excval, exctb):
 """
 Private method called to report an uncaught exception.
 def absPath(self, fn):
 """
 Public method to convert a filename to an absolute name.
 sys.path is used as a set of possible prefixes. The name stays
 relative if a file could not be found.
 @param fn filename (string)
 @return the converted filename (string)
 """
 # Eliminate anything that is part of the Python installation.
 afn = self.absPath(fn)
 for d in self.skipdirs:
 if afn.startswith(d):
 return True
 return False
 def getRunning(self):
 """
 if access:
 if oaccess:
 access = '{0!s}[{1!s}]'.format(oaccess, var[i])
 else:
 access = '{0!s}[{1!s}]'.format(access, var[i])
-if var[i-1][:-2] == '...':
+if var[i - 1][:-2] == '...':
 oaccess = access
 else:
 oaccess = ''
 try:
-loc = {"dict" : dict}
+loc = {"dict": dict}
 exec('mdict = dict{0!s}.__dict__\nobj = dict{0!s}'\
 .format(access), globals(), loc)
 mdict = loc["mdict"]
 obj = loc["obj"]
 ndict.update(mdict)
 del rvar[0:2]
 access = ""
 except:
 pass
 try:
-loc = {"cdict" : {}, "dict" : dict}
+loc = {"cdict": {}, "dict": dict}
 exec('slv = dict{0!s}.__slots__'.format(access),
 globals(), loc)
 for v in loc["slv"]:
 try:
 loc["v"] = v
 exec('cdict[v] = dict{0!s}.{1!s}'.format(access, v),
 globals, loc)
 except:
 pass
 ndict.update(loc["cdict"])
 exec('obj = dict{0!s}'.format(access), globals(), loc)
 obj = dict[var[i]]
 except:
 pass
 try:
 slv = dict[var[i]].__slots__
-loc = {"cdict" : {}, "dict" : dict, "var" : var, "i" : i}
+loc = {"cdict": {}, "dict": dict, "var": var, "i": i}
 for v in slv:
 try:
 loc["v"] = v
 exec('cdict[v] = dict[var[i]].{0!s}'.format(v),
 globals(), loc)
 except:
 pass
 ndict.update(loc["cdict"])
 obj = dict[var[i]]
 i += 1
 if ("sipThis" in dict.keys() and len(dict) == 1) or \
 (len(dict) == 0 and len(udict) > 0):
 if access:
-loc = {"udict" : udict}
+loc = {"udict": udict}
 exec('qvar = udict{0!s}'.format(access), globals(), loc)
 qvar = loc["qvar"]
 # this has to be in line with VariablesViewer.indicators
 elif rvar and rvar[0][-2:] in ["[]", "()", "{}"]:
-loc = {"udict" : udict}
+loc = {"udict": udict}
 exec('qvar = udict["{0!s}"][{1!s}]'.format(rvar[0][:-2], rvar[1]),
 globals(), loc)
 qvar = loc["qvar"]
 else:
 qvar = udict[var[-1]]
 qvtype = str(type(qvar))[1:-1].split()[1][1:-1]
 if dictkeys is None:
 dictkeys = dict.keys()
 else:
 # treatment for sequences and dictionaries
 if access:
-loc = {"dict" : dict}
+loc = {"dict": dict}
 exec("dict = dict{0!s}".format(access), globals(), loc)
 dict = loc["dict"]
 else:
 dict = dict[dictkeys[0]]
 if isDict:
 dictkeys = dict.keys()
 else:
 dictkeys = range(len(dict))
 vlist = self.__formatVariablesList(dictkeys, dict, scope, filter,
 formatSequences)
 varlist.extend(vlist)
 if obj is not None and not formatSequences:
 if repr(obj).startswith('{'):
 """
 Private method to produce a formatted output of a simple Qt4 type.
 @param value variable to be formatted
 @param vtype type of the variable to be formatted (string)
 @return A tuple consisting of a list of formatted variables. Each
 variable entry is a tuple of three elements, the variable name,
 its type and value.
 """
 qttype = vtype.split('.')[-1]
 varlist = []
 if qttype == 'QChar':
 varlist.append(("width", "float", "{0:g}".format(value.width())))
 varlist.append(("height", "float", "{0:g}".format(value.height())))
 elif qttype == 'QColor':
 varlist.append(("name", "str", "{0}".format(value.name())))
 r, g, b, a = value.getRgb()
 varlist.append(("rgb", "int",
 "{0:d}, {1:d}, {2:d}, {3:d}".format(r, g, b, a)))
 h, s, v, a = value.getHsv()
 varlist.append(("hsv", "int",
 "{0:d}, {1:d}, {2:d}, {3:d}".format(h, s, v, a)))
 c, m, y, k, a = value.getCmyk()
 varlist.append(("cmyk", "int",
 "{0:d}, {1:d}, {2:d}, {3:d}, {4:d}".format(c, m, y, k, a)))
 elif qttype == 'QDate':
 varlist.append(("", "QDate", "{0}".format(value.toString())))
 elif qttype == 'QTime':
 varlist.append(("", "QTime", "{0}".format(value.toString())))
 elif qttype == 'QDateTime':
 varlist.append(("", "QDateTime", "{0}".format(value.toString())))
 elif qttype == 'QDir':
 varlist.append(("path", "str", "{0}".format(value.path())))
 varlist.append(("absolutePath", "str",
 "{0}".format(value.absolutePath())))
 varlist.append(("canonicalPath", "str",
 "{0}".format(value.canonicalPath())))
 elif qttype == 'QFile':
 varlist.append(("fileName", "str", "{0}".format(value.fileName())))
 elif qttype == 'QFont':
 varlist.append(("family", "str", "{0}".format(value.family())))
 varlist.append(("valid", "bool", "{0}".format(value.isValid())))
 if value.isValid():
 varlist.append(("row", "int", "{0}".format(value.row())))
 varlist.append(("column", "int", "{0}".format(value.column())))
 varlist.append(("internalId", "int", "{0}".format(value.internalId())))
 varlist.append(("internalPointer", "void *",
 "{0}".format(value.internalPointer())))
 elif qttype == 'QRegExp':
 varlist.append(("pattern", "str", "{0}".format(value.pattern())))
 # GUI stuff
 varlist.append(("name", "str", "{0}".format(value.objectName())))
 varlist.append(("text", "str", "{0}".format(value.text())))
 varlist.append(("icon text", "str", "{0}".format(value.iconText())))
 varlist.append(("tooltip", "str", "{0}".format(value.toolTip())))
 varlist.append(("whatsthis", "str", "{0}".format(value.whatsThis())))
 varlist.append(("shortcut", "str",
 "{0}".format(value.shortcut().toString())))
 elif qttype == 'QKeySequence':
 varlist.append(("value", "", "{0}".format(value.toString())))
 # XML stuff
 elif qttype == 'QHostAddress':
 varlist.append(("address", "QHostAddress", "{0}".format(value.toString())))
 return varlist
-def __formatVariablesList(self, keylist, dict, scope, filter = [],
+def __formatVariablesList(self, keylist, dict, scope, filter=[],
-formatSequences = False):
+formatSequences=False):
 """
 Private method to produce a formated variables list.
 The dictionary passed in to it is scanned. Variables are
 only added to the list, if their type is not contained
 in the filter list and their name doesn't match any of the filter expressions.
 The formated variables list (a list of tuples of 3 values) is returned.
 @param keylist keys of the dictionary
 @param dict the dictionary to be scanned
 @param scope 1 to filter using the globals filter, 0 using the locals
 filter (int).
 Variables are only added to the list, if their name do not match any of the
 filter expressions.
 @param filter the indices of variable types to be filtered. Variables are
 only added to the list, if their type is not contained in the filter
 list.
 @param formatSequences flag indicating, that sequence or dictionary variables
 should be formatted. If it is 0 (or false), just the number of items contained
 in these variables is returned. (boolean)
 @return A tuple consisting of a list of formatted variables. Each variable
 entry is a tuple of three elements, the variable name, its type and
 value.
 """
 varlist = []
 if scope:
 patternFilterObjects = self.globalsFilterObjects
 def __generateFilterObjects(self, scope, filterString):
 """
 Private slot to convert a filter string to a list of filter objects.
 @param scope 1 to generate filter for global variables, 0 for local
 variables (int)
 @param filterString string of filter patterns separated by ';'
 """
 patternFilterObjects = []
 for pattern in filterString.split(';'):
 while pos >= -len(text):
 if text[pos] in completerDelims:
 if pos == -1:
 text = ''
 else:
-text = text[pos+1:]
+text = text[pos + 1:]
 break
 pos -= 1
 try:
 comp = self.complete(text, state)
 except:
 comp = None
 self.write("{0}{1}||{2}\n".format(ResponseCompletion, str(completions), text))
-def startDebugger(self, filename = None, host = None, port = None,
+def startDebugger(self, filename=None, host=None, port=None,
-enableTrace = True, exceptions = True, tracePython = False, redirect = True):
+enableTrace=True, exceptions=True, tracePython=False, redirect=True):
 """
 Public method used to start the remote debugger.
 @param filename the program to be debugged (string)
 @param host hostname of the debug server (string)
 self.dircache = []
 self.mainFrame = None
 self.inRawMode = False
 self.debugging = True
-self.attachThread(mainThread = True)
+self.attachThread(mainThread=True)
 self.mainThread.tracePython = tracePython
 # set the system exception handling function to ensure, that
 # we report on all unhandled exceptions
 sys.excepthook = self.__unhandled_exception
 # now start debugging
 if enableTrace:
 self.mainThread.set_trace()
-def startProgInDebugger(self, progargs, wd = '', host = None,
+def startProgInDebugger(self, progargs, wd='', host=None,
-port = None, exceptions = True, tracePython = False, redirect = True):
+port=None, exceptions=True, tracePython=False, redirect=True):
 """
 Public method used to start the remote debugger.
 @param progargs commandline for the program to be debugged
 (list of strings)
 @param wd working directory for the program execution (string)
 @param host hostname of the debug server (string)
 @param port portnumber of the debug server (int)
 @param exceptions flag to enable exception reporting of the IDE (boolean)
 self.passive = True
 self.write("{0}{1}|{2:d}\n".format(PassiveStartup, self.running, exceptions))
 self.__interact()
-self.attachThread(mainThread = True)
+self.attachThread(mainThread=True)
 self.mainThread.tracePython = tracePython
 # set the system exception handling function to ensure, that
 # we report on all unhandled exceptions
 sys.excepthook = self.__unhandled_exception
 @param scriptname name of the script to be debugged (string)
 @param func function to be called
 @param *args arguments being passed to func
 @return result of the function call
 """
-self.startDebugger(scriptname, enableTrace = False)
+self.startDebugger(scriptname, enableTrace=False)
 res = self.mainThread.runcall(func, *args)
 self.progTerminated(res)
 return res
 def __resolveHost(self, host):
 if not args:
 print("No program given. Aborting!")
 else:
 if not self.noencoding:
 self.__coding = self.defaultCoding
 self.startProgInDebugger(args, wd, host, port,
-exceptions = exceptions,
+exceptions=exceptions,
-tracePython = tracePython,
+tracePython=tracePython,
-redirect = redirect)
+redirect=redirect)
 else:
 if sys.argv[1] == '--no-encoding':
 self.noencoding = True
 del sys.argv[1]
 if sys.argv[1] == '':
 It prevents the debugger connections from being closed.
 @param fd file descriptor to be closed (integer)
 """
 if fd in [self.readstream.fileno(), self.writestream.fileno(),
 self.errorstream.fileno()]:
 return
 DebugClientOrigClose(fd)
 def __getSysPath(self, firstEntry):
 """
 Private slot to calculate a path list including the PYTHONPATH
 environment variable.
 @param firstEntry entry to be put first in sys.path (string)
 @return path list for use as sys.path (list of strings)
 """
 sysPath = [path for path in os.environ.get("PYTHONPATH", "").split(os.pathsep)
 if path not in sys.path] + sys.path[:]
 if "" in sysPath:
 sysPath.remove("")
 sysPath.insert(0, firstEntry)
 sysPath.insert(0, '')

Mercurial Repositories > eric / file comparison

comparison: DebugClients/Python3/DebugClientBase.py

DebugClients/Python3/DebugClientBase.py