如何创建一个像jupyter笔记本这样的具有交互式编程的App？

Question

现在，我想开发一个关于交互式编程的应用程序，就像jupyter笔记本一样，但我对这方面知之甚少，有人可以告诉我一些方法或知识来开始开发这个应用程序。

Answer 1

阶段：

1. 为您的应用程序创建虚拟环境。
2. 创建用于运行 jupyter-cells 代码的虚拟环境。
3. 在客户端-服务器的基础上实现两个虚拟环境之间的交互机制。
   服务器将要执行的单元发送到客户端。
   客户端执行单元并将结果返回到服务器。
4. 实现服务器端所需的图形界面。

两个虚拟环境之间的交互机制可以组织如下

---

server.py：

# venv-1

import sys
from multiprocessing.connection import Listener, Connection


def read_write_function(conn_for_execution: Connection, conn_for_interrupting: Connection):
    try:
        while True:
            try:
                std, received_output = conn_for_execution.recv()
            except (ConnectionResetError, KeyboardInterrupt, EOFError) as e:
                print(e)
                break
            if std in ('<stderr>', '<stdout>'):
                file = sys.stderr if std == '<stderr>' else sys.stdout
                print('stream:', std)
                print('message:', repr(received_output)[1:-1], file=file)
            elif std == '<error>':  # error
                print('error:', repr(received_output)[1:-1], file=sys.stderr)
            elif std in ('<block>', '<read>', '<readlines>'):  # next block query or read input
                print('[Ctrl+C to send code block to client]')
                lines = []
                try:
                    while True:
                        line = input(std[1:] + ' ')
                        lines.append(line)
                except (KeyboardInterrupt, EOFError):
                    conn_for_execution.send('\n'.join(lines))
                    print(('' if lines else 'nothing ') + 'sended')
                    # --------------------- <!-- only to emulate "interrupt execution"
                    if lines and lines[-1] == '#interrupt':
                        print('[SERVER] Sleep before')
                        import time
                        time.sleep(3)
                        conn_for_interrupting.send('interrupt')
                        print('[SERVER] Interrupt message sended')
                    # --------------------- --> only to emulate "interrupt execution"
                    # --------------------- <!-- only to emulate "exit"
                    if lines and lines[-1] == '#exit':
                        print('[SERVER] Sleep before')
                        import time
                        time.sleep(3)
                        conn_for_interrupting.send('exit')
                        print('[SERVER] Exit message sended')
                    # --------------------- --> only to emulate "exit"
            elif std == '<readline>':
                print('[one line to send input data to client]')
                conn_for_execution.send(input(std[1:] + ' '))
                print(std[1:] + ' sended')
    except:
        __import__('traceback').print_exc()


ADDRESS = 'localhost'
PORT = 60000
PASS = 'secret'

print('#' * 42)
print('Address:', ADDRESS)
print('Port:', PORT)
print('Pass:', PASS)
print('#' * 42)
print('Waiting for a client...')

# --------------------- <!-- only to run the client app on the server side and prevent Ctrl+C crashes

"""
import signal
import subprocess
import os


def pre_exec():
    signal.signal(signal.SIGINT, signal.SIG_IGN)  # ignore CTRL+C signal in the new process


executable = [os.path.join(os.path.abspath('ClientSide'), 'venv', 'Scripts', 'python'), '-uBq', 'client.py',
              f'--address={ADDRESS}',
              f'--port={PORT}',
              f'--password={PASS}',
              stdin=subprocess.DEVNULL]
if sys.platform.startswith('win'):
    exec_process = subprocess.Popen(executable, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
else:
    exec_process = subprocess.Popen(executable, preexec_fn=pre_exec)
"""

# --------------------- --> only to run the client app on the server side and prevent Ctrl+C crashes


# backlog = 2 --> Two clients: one for executing code blocks and one for interrupting execution
try:
    with Listener((ADDRESS, PORT), authkey=PASS.encode(encoding='utf-8'), backlog=2) as listener, \
            listener.accept() as conn_for_execution, listener.accept() as conn_for_interrupting:
        print('Connections accepted')
        print('#' * 42)
        read_write_function(conn_for_execution, conn_for_interrupting)
except:
    pass

运行：
ServerSide/venv/Scripts/python -uB server.py

---

client.py：

# venv-2

import argparse
import os
import sys
from _thread import get_native_id
from code import InteractiveInterpreter
from io import TextIOWrapper, BytesIO
from multiprocessing.connection import Client, Connection
from threading import Thread, Event

parser = argparse.ArgumentParser(prog='client.py')
parser.add_argument('--address', nargs='?', help='address ("localhost" by default)')
parser.add_argument('--port', nargs='?', help='port ("60000" by default)')
parser.add_argument('--password', nargs='?', help='password ("secret" by default)')
args = parser.parse_args()


if os.path.exists(__file__) and os.path.basename(__file__).startswith('tmp'):
    os.remove(__file__)


class Redirector(TextIOWrapper):
    def __init__(self, conn: Connection, std: TextIOWrapper):
        super().__init__(buffer=BytesIO(), encoding=std.encoding, errors=std.errors,
                         newline=std.newlines, line_buffering=std.line_buffering,
                         write_through=std.write_through)
        self.std = std
        self._conn = conn

    def read(self, size: int | None = None) -> str:
        try:
            self._conn.send(('<read>', 'read operation'))
            return self._conn.recv()
        except BaseException as e:
            print(e, file=sys.__stderr__)
            return ''

    def readline(self, size: int | None = None) -> str:
        try:
            self._conn.send(('<readline>', 'readline operation'))
            return self._conn.recv()
        except BaseException as e:
            print(e, file=sys.__stderr__)
            return ''

    def readlines(self, hint: int | None = None) -> list[str]:
        try:
            self._conn.send(('<readlines>', 'readlines operation'))
            return self._conn.recv().splitlines()
        except BaseException as e:
            print(e, file=sys.__stderr__)
            return []

    def write(self, data):
        try:
            self._conn.send((self.std.name, data))
        except BaseException as e:
            print(e, file=sys.__stderr__)

    def writelines(self, lines: list[str]):
        try:
            self._conn.send((self.std.name, os.linesep.join(lines)))
        except BaseException as e:
            print(e, file=sys.__stderr__)


class CodeBlocksInterpreter(InteractiveInterpreter):
    def __init__(self, conn_for_execution: Connection, conn_for_interrupting: Connection, locals: dict = None):
        super().__init__()
        self.locals = locals
        self._conn_for_execution = conn_for_execution
        self._conn_for_interrupting = conn_for_interrupting
        self._main_thread_id = get_native_id()
        self._ready_for_next_block = Event()
        self._ready_for_next_block.clear()
        self._can_interrupt = Event()
        self._can_interrupt.clear()
        self._thread = Thread(target=self._stop_and_exit_thread, daemon=False)

    def interact(self):
        self._thread.start()
        try:
            filename = '<input>'
            symbol = 'exec'
            while True:
                self._can_interrupt.clear()
                self._ready_for_next_block.wait()
                try:
                    self._conn_for_execution.send(('<block>', 'give me next block'))
                    code_block = self._conn_for_execution.recv() + '\n'
                    code = self.compile(source=code_block, filename=filename, symbol=symbol)
                    if code is None:
                        self.write('EOFError. Code block is incomplete')
                        continue
                    self._can_interrupt.set()
                    self.runcode(code)
                    self._can_interrupt.clear()
                except KeyboardInterrupt as e:
                    print(e, file=sys.__stderr__)
                except (OverflowError, SyntaxError, ValueError):
                    self.showsyntaxerror(filename)
                except SystemExit:
                    break
        except BaseException as e:
            print(e, file=sys.__stderr__)
        try:
            self._conn_for_execution.close()
        except:
            pass
        try:
            self._conn_for_interrupting.close()
        except:
            pass

    def _stop_and_exit_thread(self):
        try:
            while True:
                try:
                    self._ready_for_next_block.set()
                    received = self._conn_for_interrupting.recv()
                    if received == 'interrupt':
                        self._ready_for_next_block.clear()
                        if self._can_interrupt.is_set():
                            import ctypes
                            ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(self._main_thread_id),
                                                                       ctypes.py_object(KeyboardInterrupt))
                    elif received == 'exit':
                        import ctypes
                        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(self._main_thread_id),
                                                                   ctypes.py_object(SystemExit))
                        break
                except (ConnectionResetError, EOFError):
                    break
        except BaseException as e:
            print(e, file=sys.__stderr__)

    def write(self, data: str):
        self._conn_for_execution.send(('<error>', data))


ADDRESS = args.address.strip('"\'') if isinstance(args.address, str) else 'localhost'
PORT = int(args.port) if isinstance(args.port, str) and args.port.isdigit() else 60000
PASS = args.password.strip('"\'').encode('utf-8') if isinstance(args.password, str) else b'secret'

# Two clients: one for executing code blocks and one for interrupting execution
try:
    with Client((ADDRESS, PORT), authkey=PASS) as conn_for_execution, \
            Client((ADDRESS, PORT), authkey=PASS) as conn_for_interrupting:
        sys.stdin = Redirector(conn_for_execution, sys.stdin)
        sys.stdout = Redirector(conn_for_execution, sys.stdout)
        sys.stderr = Redirector(conn_for_execution, sys.stderr)
        sys.__stdin__ = Redirector(conn_for_execution, sys.__stdin__)
        sys.__stdout__ = Redirector(conn_for_execution, sys.__stdout__)
        sys.__stderr__ = Redirector(conn_for_execution, sys.__stderr__)
        code_blocks_interpreter = CodeBlocksInterpreter(conn_for_execution, conn_for_interrupting,
                                                        locals={'__name__': '__main__'})
        code_blocks_interpreter.interact()
except:
    pass

if isinstance(sys.stdin, Redirector):
    sys.stdin = sys.stdin.std
if isinstance(sys.stdout, Redirector):
    sys.stdout = sys.stdout.std
if isinstance(sys.stderr, Redirector):
    sys.stderr = sys.stderr.std
if isinstance(sys.__stdin__, Redirector):
    sys.__stdin__ = sys.__stdin__.std
if isinstance(sys.__stdout__, Redirector):
    sys.__stdout__ = sys.__stdout__.std
if isinstance(sys.__stderr__, Redirector):
    sys.__stderr__ = sys.__stderr__.std

在之后运行server.py：
ClientSide/venv/Scripts/python -uB client.py

---

在服务器端，输入代码块并发送 Ctrl+C。
在客户端执行，并将结果传回服务器端。

示例：

1. 打印到标准输出：
   [Ctrl+C 发送代码块到客户端]
块> 打印(42 ** 42)
block>
2. 打印到 stdout 和 stderr，引发异常：
   [Ctrl+C 发送代码块到客户端]
block> 导入系统、时间
block> print('1', file=sys.stdout); time.sleep(1)
block> print('2', file=sys.stderr); time.sleep(1)
阻止> 引发异常('3')
阻止>
3. 阅读：
   [Ctrl+C 发送代码块到客户端]
阻止> 导入系统
block> s1 = sys.stdin.read()
阻止>
读取> <多行>
阅读>
block> s2 = sys.stdin.readline() （或 s2 = input()< /强>)
   阻止>
readline> <单行>
block> s3 = sys.stdin.readlines()
阻止>
readlines> <多行>
readlines>
块> 打印(s1, s2, s3)
block>
4. 中断（#interrupt 必须是最后一行代码）：< br>
   [Ctrl+C 发送代码块到客户端]
阻止> 导入时间
块> 对于范围（10）内的 i：
块> print(i)
block> time.sleep(1)
阻止> #interrupt
阻止>
[SERVER] 之前睡觉
[SERVER] 中断消息发送
5. 退出（#exit 必须是最后一行代码）：
   [Ctrl+C 发送代码块到客户端]
阻止> 导入时间
块> 对于范围（10）内的 i：
块> print(i)
block> time.sleep(1)
阻止> #exit
阻止>
[SERVER] 之前睡觉
[SERVER] 已发送退出消息