-
Notifications
You must be signed in to change notification settings - Fork 280
/
discovery.py
131 lines (104 loc) · 4.02 KB
/
discovery.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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
from abc import ABCMeta, abstractmethod
import entrypoints
import logging
import six
from .kernelspec import KernelSpecManager
from .manager import KernelManager
log = logging.getLogger(__name__)
class KernelProviderBase(six.with_metaclass(ABCMeta, object)):
id = None # Should be a short string identifying the provider class.
@abstractmethod
def find_kernels(self):
"""Return an iterator of (kernel_name, kernel_info_dict) tuples."""
pass
@abstractmethod
def make_manager(self, name):
"""Make and return a KernelManager instance to start a specified kernel
name will be one of the kernel names produced by find_kernels()
"""
pass
class KernelSpecProvider(KernelProviderBase):
"""Offers kernel types from installed kernelspec directories.
"""
id = 'spec'
def __init__(self):
self.ksm = KernelSpecManager()
def find_kernels(self):
for name, resdir in self.ksm.find_kernel_specs().items():
spec = self.ksm._get_kernel_spec_by_name(name, resdir)
yield name, {
# TODO: get full language info
'language': {'name': spec.language},
'display_name': spec.display_name,
'argv': spec.argv,
}
def make_manager(self, name):
spec = self.ksm.get_kernel_spec(name)
return KernelManager(kernel_cmd=spec.argv, extra_env=spec.env)
class IPykernelProvider(KernelProviderBase):
"""Offers a kernel type using the Python interpreter it's running in.
This checks if ipykernel is importable first.
"""
id = 'pyimport'
def _check_for_kernel(self):
try:
from ipykernel.kernelspec import RESOURCES, get_kernel_dict
from ipykernel.ipkernel import IPythonKernel
except ImportError:
return None
else:
return {
'spec': get_kernel_dict(),
'language_info': IPythonKernel.language_info,
'resources_dir': RESOURCES,
}
def find_kernels(self):
info = self._check_for_kernel()
if info:
yield 'kernel', {
'language': info['language_info'],
'display_name': info['spec']['display_name'],
'argv': info['spec']['argv'],
}
def make_manager(self, name):
info = self._check_for_kernel()
if info is None:
raise Exception("ipykernel is not importable")
return KernelManager(kernel_cmd=info['spec']['argv'])
class KernelFinder(object):
"""Manages a collection of kernel providers to find available kernel types
*providers* should be a list of kernel provider instances.
"""
def __init__(self, providers):
self.providers = providers
@classmethod
def from_entrypoints(cls):
"""Load all kernel providers advertised by entry points.
Kernel providers should use the "jupyter_client.kernel_providers"
entry point group.
Returns an instance of KernelFinder.
"""
providers = []
for ep in entrypoints.get_group_all('jupyter_client.kernel_providers'):
try:
provider = ep.load()() # Load and instantiate
except Exception:
log.error('Error loading kernel provider', exc_info=True)
else:
providers.append(provider)
return cls(providers)
def find_kernels(self):
"""Iterate over available kernel types.
Yields 2-tuples of (prefixed_name, attributes)
"""
for provider in self.providers:
for kid, attributes in provider.find_kernels():
id = provider.id + '/' + kid
yield id, attributes
def make_manager(self, name):
"""Make a KernelManager instance for a given kernel type.
"""
provider_id, kernel_id = name.split('/', 1)
for provider in self.providers:
if provider_id == provider.id:
return provider.make_manager(kernel_id)