-
-
Notifications
You must be signed in to change notification settings - Fork 358
/
comm.py
89 lines (68 loc) · 2.71 KB
/
comm.py
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
"""Base class for a Comm"""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import uuid
from typing import Optional
import comm.base_comm
import traitlets.config
from traitlets import Bool, Bytes, Instance, Unicode, default
from ipykernel.jsonutil import json_clean
from ipykernel.kernelbase import Kernel
# this is the class that will be created if we do comm.create_comm
class BaseComm(comm.base_comm.BaseComm):
"""The base class for comms."""
kernel: Optional[Kernel] = None
def publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys):
"""Helper for sending a comm message on IOPub"""
if not Kernel.initialized():
return
data = {} if data is None else data
metadata = {} if metadata is None else metadata
content = json_clean(dict(data=data, comm_id=self.comm_id, **keys))
if self.kernel is None:
self.kernel = Kernel.instance()
self.kernel.session.send(
self.kernel.iopub_socket,
msg_type,
content,
metadata=json_clean(metadata),
parent=self.kernel.get_parent("shell"),
ident=self.topic,
buffers=buffers,
)
# but for backwards compatibility, we need to inherit from LoggingConfigurable
class Comm(BaseComm, traitlets.config.LoggingConfigurable):
"""Class for communicating between a Frontend and a Kernel"""
kernel = Instance("ipykernel.kernelbase.Kernel", allow_none=True) # type:ignore[assignment]
comm_id = Unicode()
primary = Bool(True, help="Am I the primary or secondary Comm?")
target_name = Unicode("comm")
target_module = Unicode(
None,
allow_none=True,
help="""requirejs module from
which to load comm target.""",
)
topic = Bytes()
@default("kernel")
def _default_kernel(self):
if Kernel.initialized():
return Kernel.instance()
@default("comm_id")
def _default_comm_id(self):
return uuid.uuid4().hex
def __init__(self, target_name='', data=None, metadata=None, buffers=None, **kwargs):
"""Initialize a comm."""
# Handle differing arguments between base classes.
had_kernel = 'kernel' in kwargs
kernel = kwargs.pop('kernel', None)
if target_name:
kwargs['target_name'] = target_name
BaseComm.__init__(
self, data=data, metadata=metadata, buffers=buffers, **kwargs
) # type:ignore[call-arg]
# only re-add kernel if explicitly provided
if had_kernel:
kwargs['kernel'] = kernel
traitlets.config.LoggingConfigurable.__init__(self, **kwargs)
__all__ = ["Comm"]