|
1 | 1 | # coding: utf-8
|
2 | 2 |
|
3 |
| - |
| 3 | +import importlib |
4 | 4 | import os
|
5 | 5 | import webbrowser
|
| 6 | +import baseObject |
6 | 7 | import shellapi
|
| 8 | +import api |
| 9 | +import config |
| 10 | +import keyboardHandler |
| 11 | +import scriptHandler |
| 12 | +import globalCommands |
| 13 | +import vision |
7 | 14 | from abc import ABC, abstractmethod
|
| 15 | +from contextlib import contextmanager |
| 16 | +from functools import partial |
| 17 | +from copy import deepcopy |
8 | 18 | from dataclasses import dataclass
|
| 19 | +from urllib import parse |
9 | 20 | from logHandler import log
|
10 |
| -from .command_uri import CommandUri |
| 21 | + |
| 22 | + |
| 23 | +USER_COMMANDS_JSON_HEADER = ( |
| 24 | + "[\n" |
| 25 | + " {\n" |
| 26 | + ' "category": "app",\n' |
| 27 | + ' "label": "Open Calculator",\n' |
| 28 | + ' "command_info": "calc.exe"\n' |
| 29 | + " }\n" |
| 30 | + "\n" |
| 31 | + "]" |
| 32 | +) |
| 33 | + |
| 34 | + |
| 35 | +@contextmanager |
| 36 | +def cwd(): |
| 37 | + old_cwd = os.getcwd() |
| 38 | + try: |
| 39 | + home_dir = os.path.expanduser("~") |
| 40 | + os.chdir(home_dir) |
| 41 | + yield |
| 42 | + finally: |
| 43 | + os.chdir(old_cwd) |
11 | 44 |
|
12 | 45 |
|
13 | 46 | class CommandError(Exception):
|
14 | 47 | """Represent failure to execute command."""
|
15 | 48 |
|
16 | 49 |
|
| 50 | +class ChangeCommand(Exception): |
| 51 | + """Change the old command to the new command.""" |
| 52 | + |
| 53 | + def __init__(self, old_command, new_command): |
| 54 | + self.old_command = old_command |
| 55 | + self.new_command = new_command |
| 56 | + |
17 | 57 |
|
18 | 58 | @dataclass
|
19 | 59 | class CommandInterpreter(ABC):
|
20 |
| - format = None |
21 |
| - registered_runners = {} |
| 60 | + __slots__ = [ |
| 61 | + "label", |
| 62 | + "command_info", |
| 63 | + "args", |
| 64 | + ] |
| 65 | + category = None |
| 66 | + registered_categories = {} |
| 67 | + __requires_text_arg__ = False |
| 68 | + __text_entry_label__ = None |
22 | 69 |
|
23 | 70 | def __init_subclass__(cls, **kwargs):
|
24 | 71 | super().__init_subclass__(**kwargs)
|
25 |
| - cls.registered_runners[cls.format] = cls |
| 72 | + cls.registered_categories[cls.category] = cls |
| 73 | + |
| 74 | + def __init__(self, command_info, args=None, label=None): |
| 75 | + self.label = label |
| 76 | + self.command_info = command_info |
| 77 | + self.args = args or {} |
| 78 | + |
| 79 | + def __hash__(self): |
| 80 | + return hash((self.category, self.label, self.command_info)) |
26 | 81 |
|
27 |
| - def __init__(self, command, **extra_args): |
28 |
| - self.command = command |
29 |
| - self.extra_args = extra_args |
| 82 | + def __repr__(self): |
| 83 | + return f"CommandInterpreter (category='{self.category}', command_info='{self.command_info}', args={self.args})" |
| 84 | + |
| 85 | + @classmethod |
| 86 | + def create(cls, category, command_info, args=None, label=None): |
| 87 | + command_cls = cls.registered_categories[category] |
| 88 | + return command_cls(command_info, args, label) |
| 89 | + |
| 90 | + @property |
| 91 | + def requires_text_arg(self): |
| 92 | + return self.__requires_text_arg__ or self.args.get("requires_text_arg") |
| 93 | + |
| 94 | + @property |
| 95 | + def text_entry_label(self): |
| 96 | + if self.requires_text_arg: |
| 97 | + return self.__text_entry_label__ or self.args.get("text_entry_label") |
| 98 | + |
| 99 | + def create_copy(self, command_info=None, args=None, label=None): |
| 100 | + clone = deepcopy(self) |
| 101 | + clone.command_info = command_info or self.command_info |
| 102 | + clone.args.update(args or {}) |
| 103 | + clone.label = label or self.label |
| 104 | + return clone |
30 | 105 |
|
31 | 106 | @abstractmethod
|
32 | 107 | def run(self):
|
33 | 108 | """Run this command."""
|
34 | 109 |
|
35 | 110 |
|
36 |
| - |
37 |
| -def run_command_by_uri(command_uri): |
| 111 | +def run_command(command): |
38 | 112 | try:
|
39 |
| - command = CommandUri.from_uri_string(command_uri) |
40 |
| - except ValueError as e: |
41 |
| - raise CommandError("Could not parse command.") from e |
42 |
| - if not command.format in CommandInterpreter.registered_runners: |
43 |
| - raise CommandError("Command not found") |
44 |
| - interpreter_cls = CommandInterpreter.registered_runners[command.format] |
45 |
| - interpreter = interpreter_cls( |
46 |
| - command=command.path, |
47 |
| - **command.primary_args |
48 |
| - ) |
49 |
| - interpreter.run() |
50 |
| - |
51 |
| - |
52 |
| -class ShellCommandInterpreter(CommandInterpreter): |
53 |
| - format = "shell" |
| 113 | + command.run() |
| 114 | + except ChangeCommand as e: |
| 115 | + run_command(e.new_command) |
| 116 | + |
| 117 | + |
| 118 | +class ShellExecuteCommandInterpreter(CommandInterpreter): |
| 119 | + category = "app" |
54 | 120 |
|
55 | 121 | def run(self):
|
56 |
| - if self.command == "home": |
57 |
| - self.command = os.path.expanduser("~") |
58 |
| - cmd = f'"{self.command}"' |
59 |
| - shellapi.ShellExecute(None, "open", cmd, "", "", 1) |
| 122 | + cmd = f'"{self.command_info}"' |
| 123 | + with cwd(): |
| 124 | + shellapi.ShellExecute(None, "open", cmd, "", "", 1) |
60 | 125 |
|
61 | 126 |
|
62 | 127 | class UrlOpenCommand(CommandInterpreter):
|
63 |
| - format = "url" |
| 128 | + category = "web.page" |
| 129 | + |
| 130 | + def run(self): |
| 131 | + webbrowser.open_new(self.command_info) |
| 132 | + |
| 133 | + |
| 134 | +class PythonFuncionCommand(CommandInterpreter): |
| 135 | + category = "python" |
| 136 | + |
| 137 | + def run(self): |
| 138 | + module, func = self.command_info.split(":") |
| 139 | + module = importlib.import_module(module) |
| 140 | + module.func(self) |
| 141 | + |
| 142 | + |
| 143 | +class SearchWebCommand(CommandInterpreter): |
| 144 | + category = "web.search" |
| 145 | + __requires_text_arg__ = True |
| 146 | + __text_entry_label__ = _("Search term") |
| 147 | + |
| 148 | + def run(self): |
| 149 | + if self.args.get("search_as_suffix", False): |
| 150 | + quoted = parse.quote_plus(self.args["text"]) |
| 151 | + full_search_url = f"{self.command_info.strip('/')}/{quoted}" |
| 152 | + else: |
| 153 | + query = parse.urlencode({self.args["query"]: self.args["text"]}) |
| 154 | + full_search_url = f"{self.command_info.strip('?')}?{query}" |
| 155 | + raise ChangeCommand( |
| 156 | + old_command=self, new_command=UrlOpenCommand(full_search_url) |
| 157 | + ) |
| 158 | + |
| 159 | + |
| 160 | +class SpecialCommand(CommandInterpreter): |
| 161 | + category = "special" |
64 | 162 |
|
65 | 163 | def run(self):
|
66 |
| - webbrowser.open_new(self.command) |
| 164 | + func = getattr(self, f"run_{self.command_info}", None) |
| 165 | + if func is None: |
| 166 | + raise CommandError(f"Unknown special command: {self.command}") |
| 167 | + func() |
| 168 | + |
| 169 | + def run_home(self): |
| 170 | + home_dir = os.path.normpath(os.path.expanduser("~")) |
| 171 | + raise ChangeCommand( |
| 172 | + old_command=self, new_command=ShellExecuteCommandInterpreter(home_dir) |
| 173 | + ) |
| 174 | + |
| 175 | + def run_open_user_commands_json(self): |
| 176 | + from .command_store import USER_COMMANDS_JSON |
| 177 | + |
| 178 | + if not os.path.isfile(USER_COMMANDS_JSON): |
| 179 | + with open(USER_COMMANDS_JSON, "w", encoding="utf-8") as newfile: |
| 180 | + newfile.write(USER_COMMANDS_JSON_HEADER) |
| 181 | + raise ChangeCommand( |
| 182 | + old_command=self, |
| 183 | + new_command=ShellExecuteCommandInterpreter(USER_COMMANDS_JSON), |
| 184 | + ) |
| 185 | + |
| 186 | + def run_open_scratchpad_directory(self): |
| 187 | + scratchpad_directory = config.getScratchpadDir() |
| 188 | + raise ChangeCommand( |
| 189 | + old_command=self, |
| 190 | + new_command=ShellExecuteCommandInterpreter(scratchpad_directory), |
| 191 | + ) |
| 192 | + |
| 193 | + |
| 194 | +class NVDAGestureCommand(CommandInterpreter): |
| 195 | + category = "nvda" |
| 196 | + |
| 197 | + def run(self): |
| 198 | + script_func = self.findScript( |
| 199 | + module=self.command_info.moduleName, |
| 200 | + cls=self.command_info.cls, |
| 201 | + scriptName=self.command_info.scriptName, |
| 202 | + ) |
| 203 | + if script_func is None: |
| 204 | + func = getattr( |
| 205 | + self.command_info.cls, f"script_{self.command_info.scriptName}" |
| 206 | + ) |
| 207 | + script_func = partial(func, None) |
| 208 | + first_kb_gesture = tuple( |
| 209 | + filter(lambda g: g.startswith("kb:"), self.command_info.gestures) |
| 210 | + ) |
| 211 | + if first_kb_gesture: |
| 212 | + gesture = keyboardHandler.KeyboardInputGesture.fromName( |
| 213 | + first_kb_gesture[0][3:] |
| 214 | + ) |
| 215 | + scriptHandler.queueScript(script_func, gesture) |
| 216 | + else: |
| 217 | + script_func(None) |
| 218 | + |
| 219 | + def findScript(self, module, cls, scriptName): |
| 220 | + focus = api.getFocusObject() |
| 221 | + if not focus: |
| 222 | + return None |
| 223 | + if scriptName.startswith("kb:"): |
| 224 | + # Emulate a key press. |
| 225 | + return scriptHandler._makeKbEmulateScript(scriptName) |
| 226 | + # Global plugin level. |
| 227 | + if cls == "GlobalPlugin": |
| 228 | + for plugin in globalPluginHandler.runningPlugins: |
| 229 | + if module == plugin.__module__: |
| 230 | + func = getattr(plugin, "script_%s" % scriptName, None) |
| 231 | + if func: |
| 232 | + return func |
| 233 | + # App module level. |
| 234 | + app = focus.appModule |
| 235 | + if app and cls == "AppModule" and module == app.__module__: |
| 236 | + func = getattr(app, "script_%s" % scriptName, None) |
| 237 | + if func: |
| 238 | + return func |
| 239 | + # Vision enhancement provider level |
| 240 | + for provider in vision.handler.getActiveProviderInstances(): |
| 241 | + if isinstance(provider, baseObject.ScriptableObject): |
| 242 | + if cls == "VisionEnhancementProvider" and module == provider.__module__: |
| 243 | + func = getattr(app, "script_%s" % scriptName, None) |
| 244 | + if func: |
| 245 | + return func |
| 246 | + # Tree interceptor level. |
| 247 | + treeInterceptor = focus.treeInterceptor |
| 248 | + if treeInterceptor and treeInterceptor.isReady: |
| 249 | + func = getattr(treeInterceptor, "script_%s" % scriptName, None) |
| 250 | + if func: |
| 251 | + return func |
| 252 | + # NVDAObject level. |
| 253 | + func = getattr(focus, "script_%s" % scriptName, None) |
| 254 | + if func: |
| 255 | + return func |
| 256 | + for obj in reversed(api.getFocusAncestors()): |
| 257 | + func = getattr(obj, "script_%s" % scriptName, None) |
| 258 | + if func and getattr(func, "canPropagate", False): |
| 259 | + return func |
| 260 | + # Global commands. |
| 261 | + func = getattr(globalCommands.commands, "script_%s" % scriptName, None) |
| 262 | + if func: |
| 263 | + return func |
| 264 | + return None |
0 commit comments