comparison libervia/cli/cmd_shell.py @ 4075:47401850dec6

refactoring: rename `libervia.frontends.jp` to `libervia.cli`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 14:54:26 +0200
parents libervia/frontends/jp/cmd_shell.py@26b7ed2817da
children
comparison
equal deleted inserted replaced
4074:26b7ed2817da 4075:47401850dec6
1 #!/usr/bin/env python3
2
3
4 # Libervia CLI
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20
21 import cmd
22 import sys
23 import shlex
24 import subprocess
25 from . import base
26 from libervia.backend.core.i18n import _
27 from libervia.backend.core import exceptions
28 from libervia.cli.constants import Const as C
29 from libervia.cli import arg_tools
30 from libervia.backend.tools.common.ansi import ANSI as A
31
32 __commands__ = ["Shell"]
33 INTRO = _(
34 """Welcome to {app_name} shell, the Salut à Toi shell !
35
36 This enrironment helps you using several {app_name} commands with similar parameters.
37
38 To quit, just enter "quit" or press C-d.
39 Enter "help" or "?" to know what to do
40 """
41 ).format(app_name=C.APP_NAME)
42
43
44 class Shell(base.CommandBase, cmd.Cmd):
45 def __init__(self, host):
46 base.CommandBase.__init__(
47 self, host, "shell",
48 help=_("launch libervia-cli in shell (REPL) mode")
49 )
50 cmd.Cmd.__init__(self)
51
52 def parse_args(self, args):
53 """parse line arguments"""
54 return shlex.split(args, posix=True)
55
56 def update_path(self):
57 self._cur_parser = self.host.parser
58 self.help = ""
59 for idx, path_elt in enumerate(self.path):
60 try:
61 self._cur_parser = arg_tools.get_cmd_choices(path_elt, self._cur_parser)
62 except exceptions.NotFound:
63 self.disp(_("bad command path"), error=True)
64 self.path = self.path[:idx]
65 break
66 else:
67 self.help = self._cur_parser
68
69 self.prompt = A.color(C.A_PROMPT_PATH, "/".join(self.path)) + A.color(
70 C.A_PROMPT_SUF, "> "
71 )
72 try:
73 self.actions = list(arg_tools.get_cmd_choices(parser=self._cur_parser).keys())
74 except exceptions.NotFound:
75 self.actions = []
76
77 def add_parser_options(self):
78 pass
79
80 def format_args(self, args):
81 """format argument to be printed with quotes if needed"""
82 for arg in args:
83 if " " in arg:
84 yield arg_tools.escape(arg)
85 else:
86 yield arg
87
88 def run_cmd(self, args, external=False):
89 """run command and retur exit code
90
91 @param args[list[string]]: arguments of the command
92 must not include program name
93 @param external(bool): True if it's an external command (i.e. not libervia-cli)
94 @return (int): exit code (0 success, any other int failure)
95 """
96 # FIXME: we have to use subprocess
97 # and relaunch whole python for now
98 # because if host.quit() is called in D-Bus callback
99 # GLib quit the whole app without possibility to stop it
100 # didn't found a nice way to work around it so far
101 # Situation should be better when we'll move away from python-dbus
102 if self.verbose:
103 self.disp(
104 _("COMMAND {external}=> {args}").format(
105 external=_("(external) ") if external else "",
106 args=" ".join(self.format_args(args)),
107 )
108 )
109 if not external:
110 args = sys.argv[0:1] + args
111 ret_code = subprocess.call(args)
112 # XXX: below is a way to launch the command without creating a new process
113 # may be used when a solution to the aforementioned issue is there
114 # try:
115 # self.host._run(args)
116 # except SystemExit as e:
117 # ret_code = e.code
118 # except Exception as e:
119 # self.disp(A.color(C.A_FAILURE, u'command failed with an exception: {msg}'.format(msg=e)), error=True)
120 # ret_code = 1
121 # else:
122 # ret_code = 0
123
124 if ret_code != 0:
125 self.disp(
126 A.color(
127 C.A_FAILURE,
128 "command failed with an error code of {err_no}".format(
129 err_no=ret_code
130 ),
131 ),
132 error=True,
133 )
134 return ret_code
135
136 def default(self, args):
137 """called when no shell command is recognized
138
139 will launch the command with args on the line
140 (i.e. will launch do [args])
141 """
142 if args == "EOF":
143 self.do_quit("")
144 self.do_do(args)
145
146 def do_help(self, args):
147 """show help message"""
148 if not args:
149 self.disp(A.color(C.A_HEADER, _("Shell commands:")), end=' ')
150 super(Shell, self).do_help(args)
151 if not args:
152 self.disp(A.color(C.A_HEADER, _("Action commands:")))
153 help_list = self._cur_parser.format_help().split("\n\n")
154 print(("\n\n".join(help_list[1 if self.path else 2 :])))
155
156 # FIXME: debug crashes on exit and is not that useful,
157 # keeping it until refactoring, may be removed entirely then
158 # def do_debug(self, args):
159 # """launch internal debugger"""
160 # try:
161 # import ipdb as pdb
162 # except ImportError:
163 # import pdb
164 # pdb.set_trace()
165
166 def do_verbose(self, args):
167 """show verbose mode, or (de)activate it"""
168 args = self.parse_args(args)
169 if args:
170 self.verbose = C.bool(args[0])
171 self.disp(
172 _("verbose mode is {status}").format(
173 status=_("ENABLED") if self.verbose else _("DISABLED")
174 )
175 )
176
177 def do_cmd(self, args):
178 """change command path"""
179 if args == "..":
180 self.path = self.path[:-1]
181 else:
182 if not args or args[0] == "/":
183 self.path = []
184 args = "/".join(args.split())
185 for path_elt in args.split("/"):
186 path_elt = path_elt.strip()
187 if not path_elt:
188 continue
189 self.path.append(path_elt)
190 self.update_path()
191
192 def do_version(self, args):
193 """show current backend/CLI version"""
194 self.run_cmd(['--version'])
195
196 def do_shell(self, args):
197 """launch an external command (you can use ![command] too)"""
198 args = self.parse_args(args)
199 self.run_cmd(args, external=True)
200
201 def do_do(self, args):
202 """lauch a command"""
203 args = self.parse_args(args)
204 if (
205 self._not_default_profile
206 and not "-p" in args
207 and not "--profile" in args
208 and not "profile" in self.use
209 ):
210 # profile is not specified and we are not using the default profile
211 # so we need to add it in arguments to use current user profile
212 if self.verbose:
213 self.disp(
214 _("arg profile={profile} (logged profile)").format(
215 profile=self.profile
216 )
217 )
218 use = self.use.copy()
219 use["profile"] = self.profile
220 else:
221 use = self.use
222
223 # args may be modified by use_args
224 # to remove subparsers from it
225 parser_args, use_args = arg_tools.get_use_args(
226 self.host, args, use, verbose=self.verbose, parser=self._cur_parser
227 )
228 cmd_args = self.path + parser_args + use_args
229 self.run_cmd(cmd_args)
230
231 def do_use(self, args):
232 """fix an argument"""
233 args = self.parse_args(args)
234 if not args:
235 if not self.use:
236 self.disp(_("no argument in USE"))
237 else:
238 self.disp(_("arguments in USE:"))
239 for arg, value in self.use.items():
240 self.disp(
241 _(
242 A.color(
243 C.A_SUBHEADER,
244 arg,
245 A.RESET,
246 " = ",
247 arg_tools.escape(value),
248 )
249 )
250 )
251 elif len(args) != 2:
252 self.disp("bad syntax, please use:\nuse [arg] [value]", error=True)
253 else:
254 self.use[args[0]] = " ".join(args[1:])
255 if self.verbose:
256 self.disp(
257 "set {name} = {value}".format(
258 name=args[0], value=arg_tools.escape(args[1])
259 )
260 )
261
262 def do_use_clear(self, args):
263 """unset one or many argument(s) in USE, or all of them if no arg is specified"""
264 args = self.parse_args(args)
265 if not args:
266 self.use.clear()
267 else:
268 for arg in args:
269 try:
270 del self.use[arg]
271 except KeyError:
272 self.disp(
273 A.color(
274 C.A_FAILURE, _("argument {name} not found").format(name=arg)
275 ),
276 error=True,
277 )
278 else:
279 if self.verbose:
280 self.disp(_("argument {name} removed").format(name=arg))
281
282 def do_whoami(self, args):
283 """print profile currently used"""
284 self.disp(self.profile)
285
286 def do_quit(self, args):
287 """quit the shell"""
288 self.disp(_("good bye!"))
289 self.host.quit()
290
291 def do_exit(self, args):
292 """alias for quit"""
293 self.do_quit(args)
294
295 async def start(self):
296 # FIXME: "shell" is currently kept synchronous as it works well as it
297 # and it will be refactored soon.
298 default_profile = self.host.bridge.profile_name_get(C.PROF_KEY_DEFAULT)
299 self._not_default_profile = self.profile != default_profile
300 self.path = []
301 self._cur_parser = self.host.parser
302 self.use = {}
303 self.verbose = False
304 self.update_path()
305 self.cmdloop(INTRO)