File: echo_kernel.py

package info (click to toggle)
pytest-jupyter 0.10.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 240 kB
  • sloc: python: 644; sh: 23; makefile: 14
file content (69 lines) | stat: -rw-r--r-- 2,023 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
"""A simple echo kernel."""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
from __future__ import annotations

import logging
import typing

# mypy: disable-error-code="no-untyped-call"
from ipykernel.kernelapp import IPKernelApp
from ipykernel.kernelbase import Kernel


class EchoKernel(Kernel):
    """An echo kernel."""

    implementation = "Echo"
    implementation_version = "1.0"
    language = "echo"
    language_version = "0.1"
    language_info = {
        "name": "echo",
        "mimetype": "text/plain",
        "file_extension": ".txt",
    }
    banner = "Echo kernel - as useful as a parrot"

    def do_execute(  # type:ignore[override]
        self,
        code: str,
        silent: bool,
        store_history=True,  # noqa: ARG002
        user_expressions: typing.Any = None,  # noqa: ARG002
        allow_stdin=False,
        *,
        cell_id: str | None = None,  # noqa: ARG002
    ) -> dict[str, typing.Any]:
        """Execute code on the kernel."""
        if not silent:
            stream_content = {"name": "stdout", "text": code}
            self.send_response(self.iopub_socket, "stream", stream_content)

            # Send a input_request if code contains input command.
            if allow_stdin and code and code.find("input(") != -1:
                self._input_request(
                    "Echo Prompt",
                    self._parent_ident["shell"],
                    self.get_parent(channel="shell"),
                    password=False,
                )

        return {
            "status": "ok",
            # The base class increments the execution count
            "execution_count": self.execution_count,
            "payload": [],
            "user_expressions": {},
        }


class EchoKernelApp(IPKernelApp):
    """An app for the echo kernel."""

    kernel_class = EchoKernel  # type:ignore[assignment]


if __name__ == "__main__":
    logging.disable(logging.ERROR)
    EchoKernelApp.launch_instance()