1  """Connection file-related utilities for the kernel
   2  """
   3  # Copyright (c) IPython Development Team.
   4  # Distributed under the terms of the Modified BSD License.
   5  from __future__ import annotations
   6  
   7  import json
   8  import sys
   9  from subprocess import PIPE, Popen
  10  from typing import TYPE_CHECKING, Any
  11  
  12  import jupyter_client
  13  from jupyter_client import write_connection_file
  14  
  15  if TYPE_CHECKING:
  16      from ipykernel.kernelapp import IPKernelApp
  17  
  18  
  19  def get_connection_file(app: IPKernelApp | None = None) -> str:
  20      """Return the path to the connection file of an app
  21  
  22      Parameters
  23      ----------
  24      app : IPKernelApp instance [optional]
  25          If unspecified, the currently running app will be used
  26      """
  27      from traitlets.utils import filefind
  28  
  29      if app is None:
  30          from ipykernel.kernelapp import IPKernelApp
  31  
  32          if not IPKernelApp.initialized():
  33              msg = "app not specified, and not in a running Kernel"
  34              raise RuntimeError(msg)
  35  
  36          app = IPKernelApp.instance()
  37      return filefind(app.connection_file, [".", app.connection_dir])
  38  
  39  
  40  def _find_connection_file(connection_file):
  41      """Return the absolute path for a connection file
  42  
  43      - If nothing specified, return current Kernel's connection file
  44      - Otherwise, call jupyter_client.find_connection_file
  45      """
  46      if connection_file is None:
  47          # get connection file from current kernel
  48          return get_connection_file()
  49      return jupyter_client.find_connection_file(connection_file)
  50  
  51  
  52  def get_connection_info(
  53      connection_file: str | None = None, unpack: bool = False
  54  ) -> str | dict[str, Any]:
  55      """Return the connection information for the current Kernel.
  56  
  57      Parameters
  58      ----------
  59      connection_file : str [optional]
  60          The connection file to be used. Can be given by absolute path, or
  61          IPython will search in the security directory.
  62          If run from IPython,
  63  
  64          If unspecified, the connection file for the currently running
  65          IPython Kernel will be used, which is only allowed from inside a kernel.
  66  
  67      unpack : bool [default: False]
  68          if True, return the unpacked dict, otherwise just the string contents
  69          of the file.
  70  
  71      Returns
  72      -------
  73      The connection dictionary of the current kernel, as string or dict,
  74      depending on `unpack`.
  75      """
  76      cf = _find_connection_file(connection_file)
  77  
  78      with open(cf) as f:
  79          info_str = f.read()
  80  
  81      if unpack:
  82          info = json.loads(info_str)
  83          # ensure key is bytes:
  84          info["key"] = info.get("key", "").encode()
  85          return info  # type:ignore[no-any-return]
  86  
  87      return info_str
  88  
  89  
  90  def connect_qtconsole(
  91      connection_file: str | None = None, argv: list[str] | None = None
  92  ) -> Popen[Any]:
  93      """Connect a qtconsole to the current kernel.
  94  
  95      This is useful for connecting a second qtconsole to a kernel, or to a
  96      local notebook.
  97  
  98      Parameters
  99      ----------
 100      connection_file : str [optional]
 101          The connection file to be used. Can be given by absolute path, or
 102          IPython will search in the security directory.
 103          If run from IPython,
 104  
 105          If unspecified, the connection file for the currently running
 106          IPython Kernel will be used, which is only allowed from inside a kernel.
 107  
 108      argv : list [optional]
 109          Any extra args to be passed to the console.
 110  
 111      Returns
 112      -------
 113      :class:`subprocess.Popen` instance running the qtconsole frontend
 114      """
 115      argv = [] if argv is None else argv
 116  
 117      cf = _find_connection_file(connection_file)
 118  
 119      cmd = ";".join(["from qtconsole import qtconsoleapp", "qtconsoleapp.main()"])
 120  
 121      kwargs: dict[str, Any] = {}
 122      # Launch the Qt console in a separate session & process group, so
 123      # interrupting the kernel doesn't kill it.
 124      kwargs["start_new_session"] = True
 125  
 126      return Popen(
 127          [sys.executable, "-c", cmd, "--existing", cf, *argv],
 128          stdout=PIPE,
 129          stderr=PIPE,
 130          close_fds=(sys.platform != "win32"),
 131          **kwargs,
 132      )
 133  
 134  
 135  __all__ = [
 136      "write_connection_file",
 137      "get_connection_file",
 138      "get_connection_info",
 139      "connect_qtconsole",
 140  ]
 141