如何在 Jupyter 内核中获取单元格的 ID？

Question

我正在尝试为不真正支持 REPL 的语言构建 Jupyter 内核，并且重新定义变量或函数会在该语言中引发错误。不幸的是，这意味着我不能只按照用户提交的顺序执行代码，而是需要在他们重新访问旧单元格时替换它。假设用户有以下两个单元格：

单元格 1：

int foo = 1;

单元格 2：

vec4(foo);

在我的理想情况下，我只想将单元格拼接成一个按单元格顺序排列的虚拟源文件，然后执行该文件。所以生成的虚拟源文件应该是这样的：

int foo = 1;
vec4(foo);

现在假设用户返回单元格 1 并将 foo 编辑为 4，我如何才能知道用户编辑了单元格 1？所以理想情况下，我想将虚拟源文件更新为如下所示：

int foo = 4;
vec4(foo);

而不是这个：

int foo = 1;
vec4(foo);
int foo = 4; // This would throw an error in the language compiler

我使用this 作为我的基础，我查看了源代码，但找不到任何对我有帮助的东西。有什么我错过的吗？还有什么我应该做的吗？

Answer 1

有一个使用 messing_api (https://jupyter-client.readthedocs.io/en/latest/messaging.html#history) 的可能解决方案。

import asyncio
import os
from uuid import uuid4
import json
from dataclasses import dataclass

from tornado.escape import json_encode, json_decode, url_escape
from tornado.websocket import websocket_connect
from tornado.httpclient import AsyncHTTPClient, HTTPRequest

client = AsyncHTTPClient()
session_id = 'faf69f76-6667-45d6-a38f-32460e5d7f24'
token = 'e9e267d0c802017c22bc31d276b675b4f5b3e0f180eb5c8b'
kernel_id = 'fad149a5-1f78-4827-ba7c-f1fde844f0b2'

@dataclass
class Cell:
    code: str
    index: int
    execution_count: int

# We keep track of all cells to matain an updated index
cells = []

async def get_sessions():
    url = 'http://localhost:8888/api/sessions?token={}'.format(token)
    req = HTTPRequest(url=url)
    resp = await client.fetch(req)
    print(resp)
    print(resp.body)

async def get_notebook_content(path):
    url = 'http://localhost:8888/api/contents/{}?token={}'.format(path, token)

    req = HTTPRequest(url=url)
    resp = await client.fetch(req)
    return json_decode(resp.body)

async def get_session(session_id):
    ses_url = 'http://localhost:8888/api/sessions/{}?token={}'.format(session_id, token)
    ses_req = HTTPRequest(url=ses_url)
    resp = await client.fetch(ses_req)
    return json_decode(resp.body)

# return the list of notebook cells as Cell @dataclass
def parse_cells(content):
    res = []
    # we iterate over notebook cells
    cells = content['content']['cells']
    # search the cell
    for index, c in enumerate(cells):
        cell_execution_count = c['execution_count']
        code = c['source']
        cell = Cell(code=code, index=index, execution_count=cell_execution_count)

        res.append(cell)

    return res

# listen to all notebook messages
async def listen():
    session_data = await get_session(session_id)
    notebook_path = session_data['notebook']['path']
    notebook_content = await get_notebook_content(notebook_path)

    # parse existing cells
    cells = parse_cells(notebook_content)

    # listen to all messages
    req = HTTPRequest(
        url='ws://localhost:8888/api/kernels/{}/channels?token={}'.format(
            kernel_id,
            token))
    ws = await websocket_connect(req)
    print('Connected to kernel websocket')
    hist_msg_id = None

    while True:
        msg = await ws.read_message()
        msg = json_decode(msg)
        msg_type = msg['msg_type']
        parent_msg_id = msg['parent_header']['msg_id']

        if msg_type == 'execute_input':
            # after a executed cell we request the history (only of the last executed cell)
            hist_msg_id = uuid4().hex
            ws.write_message(json_encode({
                'header': {
                    'username': '',
                    'version': '5.3',
                    'session': '',
                    'msg_id': hist_msg_id,
                    'msg_type': 'history_request'
                },
                'parent_header': {},
                'channel': 'shell',
                'content': {
                    'output': False,
                    'raw': True,
                    'hist_access_type': 'tail',
                    'n': 1
                },
                'metadata': {
                },
                'buffers': {}
            }))
        elif parent_msg_id == hist_msg_id and msg_type == 'history_reply':
            # we receive the history of the last executed cell with his execution_count
            hist_msg_id = None # we dont expect more replies
            # see message type 'history_result': https://jupyter-client.readthedocs.io/en/latest/messaging.html#history
            execution_count = msg['content']['history'][0][1]
            code = msg['content']['history'][0][2]
            # update the existing cell
            for c in cells:
                if c.execution_count + 1 == execution_count:
                    c.code = code
                    c.execution_count = execution_count

                    print('# Cell changed: {}'.format(c))

if __name__ == '__main__':
    asyncio.run(listen())

让我试着解释一下……

• 我们在 Cell 数据类的列表（单元格）中跟踪所有笔记本单元格及其索引（代码、索引和执行计数）

• 我们监听来自所需会话的每条消息（方法监听）

• 当一个单元格被执行时，我们通过消息 api 请求他的历史，以获取代码和 execution_count

• 我们通过他的 execution_count 将单元格与现有的单元格匹配并更新它

我知道这是一个非常奇特的解决方案，但是当笔记本与消息 api 通信时，不包含有关某种单元身份的任何信息，只有他的代码。

重要提示

此解决方案不管理插入或删除单元格，我们必须使用内核历史记录或类似的东西找到解决方案...