Write a Python Debugger

# annoying_sibling_server.py

import sys
import socket


class AnnoyingSibling:
    """Makes a file stream annoying.

    Wraps a file-like object so that its write method outputs text
    with alternating caps.

    Parameters
    ----------

    file_stream : file object associated with a socket

      This should be the result of socket.makefile, the client
      connection socket.

    """

    def __init__(self, file_stream):
        self.file_stream = file_stream

    def write(self, s):
        annoying_s = ''

        for i, c in enumerate(s):
            annoying_s += c.upper() if i % 2 else c

        self.file_stream.write(annoying_s)

    def __getattr__(self, attr):
        # check all attribute lookups for a write call and dispatch
        # accordingly
        if attr == 'write':
            return self.write
        elif hasattr(self.file_stream, attr):
            return getattr(self.file_stream, attr)


if __name__ == '__main__':
    # create a TCP socket (SOCK_STREAM) which communicates using ipv4 (AF_INET)
    listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # don't hog the port when done; allow new connections
    listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
    listen_socket.bind(('127.0.0.1', 4444))
    listen_socket.listen(0)

    # create a socket to handle a client connection
    (connection_socket, address) = listen_socket.accept()
    # represent the socket using the file API
    file_stream = connection_socket.makefile('rw', errors='ignore')
    # wrap the steam's 'write' method to make it annoying
    annoying_sibling_stream = AnnoyingSibling(file_stream)

    # start the REPL
    user_input = None
    while not user_input == "exit":
        user_input = annoying_sibling_stream.readline().strip('\n')
        annoying_sibling_stream.write("%s\n" % user_input)
        annoying_sibling_stream.flush()

    # clean up on exit
    file_stream.close()
    listen_socket.shutdown(socket.SHUT_RDWR)
    listen_socket.close()

def cmdloop(self, intro):
    stop = None
    while not stop:
        if self.cmdqueue:
            line = self.cmdqueue.pop(0)
        else:
            if self.use_rawinput:
                try:
                    line = input(self.prompt)
                except EOFError:
                    line = 'EOF'
            else:
                self.stdout.write(self.prompt)
                self.stdout.flush()
                line = self.stdin.readline()
                if not len(line):
                    line = 'EOF'
                else:
                    line = line.rstrip('\r\n')
        line = self.precmd(line)
        stop = self.onecmd(line)
        stop = self.postcmd(stop, line)

# my_simple_remote_debugger.py

import sys
import socket
from pdb import Pdb


class MySimpleRemoteDebugger(Pdb):

    def __init__(self):

        self.listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
        self.listen_socket.bind(('127.0.0.1', 4444))
        self.listen_socket.listen(1)

        connection, address = self.listen_socket.accept()
        self.stream = connection.makefile('rw', errors='ignore')

        self._old_stdin_fd_or_file  = sys.stdin
        self._old_stdout_fd_or_file = sys.stdout
        self._old_stderr_fd_or_file = sys.stderr

        sys.stdin  = self.stream
        sys.stdout = self.stream
        sys.stderr = self.stream

        Pdb.__init__(self)

        # use stdin
        self.use_rawinput = False

        self.prompt = f'(MySimpleRemoteDebugger) '

    def __del__(self):
        sys.stdin = self._old_stdin_fd_or_file
        sys.stdout = self._old_stdout_fd_or_file
        sys.stderr = self._old_stderr_fd_or_file

        self.stream.close()
        self.listen_socket.shutdown(socket.SHUT_RDWR)
        self.listen_socket.close()


def breakpoint():
    DEBUGGER = MySimpleRemoteDebugger()
    DEBUGGER.frame = sys._getframe().f_back
    DEBUGGER.set_trace(DEBUGGER.frame)

 1: # my_first_remote_debugging_experience.py
 2: 
 3: import my_simple_remote_debugger
 4: 
 5: def foo(bar):
 6:     x = 1
 7:     print(bar)
 8: 
 9: 
10: def main():
11:     print("in main")
12:     my_simple_remote_debugger.breakpoint()
13:     foo('baz')
14:     print("exiting main")
15: 
16: 
17: if __name__ == '__main__':
18:     main()

# If a file .pdbrc exists in the user\342\200\231s home directory or in the
# current directory, it is read with 'utf-8' encoding and executed as
# if it had been typed at the debugger prompt.

# Show context on startup
l

# Use capital letter to do the action and then print the listing
alias N 'Stepping over...'      ;; next     ;; l
alias S 'Stepping into...'      ;; step     ;; l
alias R 'Returning...'          ;; return   ;; l
alias C 'Continuing...'         ;; continue ;; l
alias U 'Going up a frame...'   ;; up       ;; l
alias D 'Going down a frame...' ;; down     ;; l

import sys
import socket

try:
    from IPython.core.debugger import Pdb
    USING_IPYTHON = True
except ImportError:
    from pdb import Pdb
    USING_IPYTHON = False


class MyRemoteDebugger(Pdb):

    def __init__(self):

        ...

        if USING_IPYTHON:
            Pdb.__init__(self, context=10)
        else:
            Pdb.__init__(self)

        ...

# my_simple_remote_debugger.py

import sys
import socket
from pdb import Pdb


class MySimpleRemoteDebugger(Pdb):

    def __init__(self):

        # forces one connection per port by closing socket
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            # make freed port immediately available
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
            sock.bind(('127.0.0.1', 4444))
            sock.listen(0)

            connection, address = sock.accept()
            stream = connection.makefile('rw', errors='ignore')

            Pdb.__init__(self, stdin=stream, stdout=stream)

        # don't use readline for input (makes friendly for clients)
        self.use_rawinput = False
        self.prompt       = f'(MySimpleRemoteDebugger) '


def breakpoint():
    my_debugger = MySimpleRemoteDebugger()
    my_debugger.frame = sys._getframe().f_back
    my_debugger.set_trace(my_debugger.frame)

"""The Remote Python Debugger Rdb
==============================

Platform-independent remote Python debugger based on Pdb.  See
https://docs.python.org/3.12/library/pdb.html and
https://github.com/python/cpython/blob/3.12/Lib/pdb.py

Place 'rdb.py' in the local site-packages directory to make it
available to projects.

    $ python -m site --user-site
    /home/ahab/.local/lib/python3.12/site-packages/rdb.py

See https://peps.python.org/pep-0370/https://peps.python.org/pep-0370/

Set a breakpoint using

    >>> import rdb; rdb.set_trace()

or

    >>> import rdb; rdb.breakpoint()

The default host and port is 127.0.0.1:4444.

Post-mortem debugging is also available:

    def my_exception_hook(type, value, tb, thread=None):
        import rdb; rdb.pm(tb)

    import sys
    sys.excepthook = my_exception_hook

    import threading
    threading.excepthook = my_exception_hook

Rdb uses the builtin 'pdb' unless IPython 'ipdb' is installed.

Suggested ".pdbrc" aliases (home or current directory):

    # Show context on startup
    l

    alias N 'Stepping over...'      ;; next     ;; l
    alias S 'Stepping into...'      ;; step     ;; l
    alias R 'Returning...'          ;; return   ;; l
    alias C 'Continuing...'         ;; continue ;; l
    alias U 'Going up a frame...'   ;; up       ;; l
    alias D 'Going down a frame...' ;; down     ;; l

The following defines an Emacs-based client debugger for an Rdb
breakpoint.  Call M-x my-start-debugger-client.

    (defun my-start-debugger-client (&optional arg host port)
      "Connect to Rdb debugger at HOST on PORT.

    Prompt for HOST when called with universal prefix C-u.
    Otherwise, use \"127.0.0.1\"."
      (interactive "p")
      (let* ((host
              (cond ((= arg 4)
                     (read-string "host (default 127.0.0.1): " "127.0.0.1" nil "127.0.0.1"))
                    (t
                     "127.0.0.1")))
             (port (or port (read-number "port: " 4444)))
             (proc-name (format "my-debugger-client-%s-%s" host port))
             (proc-buffer (format "*%s*" proc-name)))

        (make-network-process
         :name proc-name
         :buffer proc-buffer
         :family 'ipv4
         :host host
         :service port
         :sentinel 'my--debugger-client-sentinel
         :filter 'my--debugger-client-filter)

        (switch-to-buffer proc-buffer)

        (with-current-buffer proc-buffer
          (comint-mode)
          (setq-local comint-prompt-regexp "\\((Rdb) \\|>>> \\|In \\[[[:digit:]]\\]: \\)")
          (setq-local comint-use-prompt-regexp t))))

    (defun my-stop-debugger-client nil
      (interactive)
      (delete-process "my-debugger-client"))

    (defun my--debugger-client-filter (proc string)
      (comint-output-filter proc string))

    (defun my--debugger-client-sentinel (proc msg)
      (let ((proc-buffer (format "*%s*" proc)))
        (when (string= msg "connection broken by remote peer\n")
          (with-current-buffer proc-buffer
            (insert (format "client %s has quit" proc))
            (bury-buffer)  ; keep output on close
            ;; (kill-buffer)  ; remove all output on close
            ))))

"""

import os
import sys
import socket

try:
    from IPython.core.debugger import Pdb
except ImportError:
    from pdb import Pdb


DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 4444


class Rdb(Pdb):

    def __init__(self, name=None, host=DEFAULT_HOST, port=DEFAULT_PORT, *args, **kwargs):

        if not name:
            name = f"pid:{os.getpid()}"

        # forces one connection per port by closing socket
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            # make freed port immediately available
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

            try:
                sock.bind((host, port))  # default or user provided port
            except OSError:
                sock.bind((host, 0))     # OS decides port

            sock.listen(0)               # number of unaccepted connections system allows before refusing new connections

            bound_host, bound_port = sock.getsockname()
            sys.stdout.write(f"[{name}] Starting remote debugger {bound_host}:{bound_port}...\n")

            connection, address = sock.accept()
            stream = connection.makefile('rw', errors='ignore')

            super().__init__(stdin=stream, stdout=stream, *args, **kwargs)

        # make client friendly, don't use readline (e.g. telnet, PuTTY)
        # See: https://docs.python.org/3.12/library/cmd.html#cmd.Cmd.use_rawinput
        self.use_rawinput = False
        self.prompt       = f"({self.__class__.__name__}) "


def breakpoint(host=DEFAULT_HOST, port=DEFAULT_PORT, name=None):
    remote_debugger = Rdb(name=name, host=host, port=port)
    remote_debugger.set_trace(sys._getframe().f_back)

set_trace = breakpoint


def post_mortem(t=None):
    """Enter post-mortem debugging of the given *traceback* object.

    If no traceback is given, it uses the one of the exception that is
    currently being handled (an exception must be being handled if the
    default is to be used).

    Modified from Python3.12 pdb.post_mortem
    """
    # handling the default
    if t is None:
        exc = sys.exception()
        if exc is not None:
            t = exc.__traceback__

    if t is None:
        raise ValueError("A valid traceback must be passed if no "
                         "exception is being handled")

    r = Rdb(name=f"pid:{os.getpid()}:post_mortem")
    r.reset()
    r.interaction(None, t)


def pm(tb=None):
    """Enter post-mortem debugging of the traceback found in sys.last_traceback.

    Modified from Python3.12 pdb.pm
    """
    if not tb:
        if hasattr(sys, 'last_exc'):
            tb = sys.last_exc.__traceback__
        else:
            tb = sys.last_traceback
    post_mortem(tb)

(defun my-start-debugger-client (&optional arg host port)
  "Connect to Rdb debugger at HOST on PORT.

    Prompt for HOST when called with universal prefix C-u.
    Otherwise, use \"127.0.0.1\"."
  (interactive "p")
  (let* ((host
          (cond ((= arg 4)
                 (read-string "host (default 127.0.0.1): " "127.0.0.1" nil "127.0.0.1"))
                (t
                 "127.0.0.1")))
         (port (or port (read-number "port: " 4444)))
         (proc-name (format "my-debugger-client-%s-%s" host port))
         (proc-buffer (format "*%s*" proc-name)))

    (make-network-process
     :name proc-name
     :buffer proc-buffer
     :family 'ipv4
     :host host
     :service port
     :sentinel 'my--debugger-client-sentinel
     :filter 'my--debugger-client-filter)

    (switch-to-buffer proc-buffer)

    (with-current-buffer proc-buffer
      (comint-mode)
      (setq-local comint-prompt-regexp "\\((Rdb) \\|>>> \\|In \\[[[:digit:]]\\]: \\)")
      (setq-local comint-use-prompt-regexp t))))

(defun my-stop-debugger-client nil
  (interactive)
  (delete-process "my-debugger-client"))

(defun my--debugger-client-filter (proc string)
  (comint-output-filter proc string))

(defun my--debugger-client-sentinel (proc msg)
  (let ((proc-buffer (format "*%s*" proc)))
    (when (string= msg "connection broken by remote peer\n")
      (with-current-buffer proc-buffer
        (insert (format "client %s has quit" proc))
        (bury-buffer)  ; keep output on close
        ;; (kill-buffer)  ; remove all output on close
        ))))