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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
#------------------------------------------------------------------------------
# Copyright (c) 2013-2025, Nucleic Development Team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
#------------------------------------------------------------------------------
from collections import defaultdict
from atom.api import Typed
from enaml.workbench.plugin import Plugin
from .command import Command
from .execution_event import ExecutionEvent
COMMANDS_POINT = 'enaml.workbench.core.commands'
class CorePlugin(Plugin):
""" The core plugin for the Enaml workbench.
"""
def start(self):
""" Start the plugin life-cycle.
This method is called by the framework at the appropriate time.
It should never be called by user code.
"""
self._refresh_commands()
self._bind_observers()
def stop(self):
""" Stop the plugin life-cycle.
This method is called by the framework at the appropriate time.
It should never be called by user code.
"""
self._unbind_observers()
self._commands.clear()
self._command_extensions.clear()
def invoke_command(self, command_id, parameters={}, trigger=None):
""" Invoke the command handler for the given command id.
Parameters
----------
command_id : unicode
The unique identifier of the command to invoke.
parameters : dict, optional
The parameters to pass to the command handler.
trigger : object, optional
The object which triggered the command.
Returns
-------
result : object
The return value of the command handler.
"""
if command_id not in self._commands:
msg = "'%s' is not a registered command id"
raise ValueError(msg % command_id)
command = self._commands[command_id]
event = ExecutionEvent()
event.command = command
event.workbench = self.workbench
event.parameters = parameters # copied on assignment
event.trigger = trigger
return command.handler(event)
#--------------------------------------------------------------------------
# Private API
#--------------------------------------------------------------------------
#: The mapping of command id to Command object.
_commands = Typed(dict, ())
#: The mapping of extension object to list of Command objects.
_command_extensions = Typed(defaultdict, (list,))
def _refresh_commands(self):
""" Refresh the command objects for the plugin.
"""
workbench = self.workbench
point = workbench.get_extension_point(COMMANDS_POINT)
extensions = point.extensions
if not extensions:
self._commands.clear()
self._command_extensions.clear()
return
new_extensions = defaultdict(list)
old_extensions = self._command_extensions
for extension in extensions:
if extension in old_extensions:
commands = old_extensions[extension]
else:
commands = self._load_commands(extension)
new_extensions[extension].extend(commands)
commands = {}
for extension in extensions:
for command in new_extensions[extension]:
if command.id in commands:
msg = "command '%s' is already registered"
raise ValueError(msg % command.id)
if command.handler is None:
msg = "command '%s' does not declare a handler"
raise ValueError(msg % command.id)
commands[command.id] = command
self._commands = commands
self._command_extensions = new_extensions
def _load_commands(self, extension):
""" Load the command objects for the given extension.
Parameters
----------
extension : Extension
The extension object of interest.
Returns
-------
result : list
The list of Command objects declared by the extension.
"""
workbench = self.workbench
commands = extension.get_children(Command)
if extension.factory is not None:
for item in extension.factory(workbench):
if not isinstance(item, Command):
msg = "extension '%s' created non-Command of type '%s'"
args = (extension.qualified_id, type(item).__name__)
raise TypeError(msg % args)
commands.append(item)
return commands
def _on_commands_updated(self, change):
""" The observer for the commands extension point.
"""
self._refresh_commands()
def _bind_observers(self):
""" Setup the observers for the plugin.
"""
workbench = self.workbench
point = workbench.get_extension_point(COMMANDS_POINT)
point.observe('extensions', self._on_commands_updated)
def _unbind_observers(self):
""" Remove the observers for the plugin.
"""
workbench = self.workbench
point = workbench.get_extension_point(COMMANDS_POINT)
point.unobserve('extensions', self._on_commands_updated)
|
