Mercurial > libervia-backend

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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_debug.py@26b7ed2817da
children 4ed8de94c926
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 from . import base
22 from libervia.backend.core.i18n import _
23 from libervia.cli.constants import Const as C
24 from libervia.backend.tools.common.ansi import ANSI as A
25 import json
26
27 __commands__ = ["Debug"]
28
29
30 class BridgeCommon(object):
31 def eval_args(self):
32 if self.args.arg:
33 try:
34 return eval("[{}]".format(",".join(self.args.arg)))
35 except SyntaxError as e:
36 self.disp(
37 "Can't evaluate arguments: {mess}\n{text}\n{offset}^".format(
38 mess=e, text=e.text, offset=" " * (e.offset - 1)
39 ),
40 error=True,
41 )
42 self.host.quit(C.EXIT_BAD_ARG)
43 else:
44 return []
45
46
47 class Method(base.CommandBase, BridgeCommon):
48 def __init__(self, host):
49 base.CommandBase.__init__(self, host, "method", help=_("call a bridge method"))
50 BridgeCommon.__init__(self)
51
52 def add_parser_options(self):
53 self.parser.add_argument(
54 "method", type=str, help=_("name of the method to execute")
55 )
56 self.parser.add_argument("arg", nargs="*", help=_("argument of the method"))
57
58 async def start(self):
59 method = getattr(self.host.bridge, self.args.method)
60 import inspect
61
62 argspec = inspect.getargspec(method)
63
64 kwargs = {}
65 if "profile_key" in argspec.args:
66 kwargs["profile_key"] = self.profile
67 elif "profile" in argspec.args:
68 kwargs["profile"] = self.profile
69
70 args = self.eval_args()
71
72 try:
73 ret = await method(
74 *args,
75 **kwargs,
76 )
77 except Exception as e:
78 self.disp(
79 _("Error while executing {method}: {e}").format(
80 method=self.args.method, e=e
81 ),
82 error=True,
83 )
84 self.host.quit(C.EXIT_ERROR)
85 else:
86 if ret is not None:
87 self.disp(str(ret))
88 self.host.quit()
89
90
91 class Signal(base.CommandBase, BridgeCommon):
92 def __init__(self, host):
93 base.CommandBase.__init__(
94 self, host, "signal", help=_("send a fake signal from backend")
95 )
96 BridgeCommon.__init__(self)
97
98 def add_parser_options(self):
99 self.parser.add_argument("signal", type=str, help=_("name of the signal to send"))
100 self.parser.add_argument("arg", nargs="*", help=_("argument of the signal"))
101
102 async def start(self):
103 args = self.eval_args()
104 json_args = json.dumps(args)
105 # XXX: we use self.args.profile and not self.profile
106 # because we want the raw profile_key (so plugin handle C.PROF_KEY_NONE)
107 try:
108 await self.host.bridge.debug_signal_fake(
109 self.args.signal, json_args, self.args.profile
110 )
111 except Exception as e:
112 self.disp(_("Can't send fake signal: {e}").format(e=e), error=True)
113 self.host.quit(C.EXIT_ERROR)
114 else:
115 self.host.quit()
116
117
118 class bridge(base.CommandBase):
119 subcommands = (Method, Signal)
120
121 def __init__(self, host):
122 super(bridge, self).__init__(
123 host, "bridge", use_profile=False, help=_("bridge s(t)imulation")
124 )
125
126
127 class Monitor(base.CommandBase):
128 def __init__(self, host):
129 super(Monitor, self).__init__(
130 host,
131 "monitor",
132 use_verbose=True,
133 use_profile=False,
134 use_output=C.OUTPUT_XML,
135 help=_("monitor XML stream"),
136 )
137
138 def add_parser_options(self):
139 self.parser.add_argument(
140 "-d",
141 "--direction",
142 choices=("in", "out", "both"),
143 default="both",
144 help=_("stream direction filter"),
145 )
146
147 async def print_xml(self, direction, xml_data, profile):
148 if self.args.direction == "in" and direction != "IN":
149 return
150 if self.args.direction == "out" and direction != "OUT":
151 return
152 verbosity = self.host.verbosity
153 if not xml_data.strip():
154 if verbosity <= 2:
155 return
156 whiteping = True
157 else:
158 whiteping = False
159
160 if verbosity:
161 profile_disp = f" ({profile})" if verbosity > 1 else ""
162 if direction == "IN":
163 self.disp(
164 A.color(
165 A.BOLD, A.FG_YELLOW, "<<<===== IN ====", A.FG_WHITE, profile_disp
166 )
167 )
168 else:
169 self.disp(
170 A.color(
171 A.BOLD, A.FG_CYAN, "==== OUT ====>>>", A.FG_WHITE, profile_disp
172 )
173 )
174 if whiteping:
175 self.disp("[WHITESPACE PING]")
176 else:
177 try:
178 await self.output(xml_data)
179 except Exception:
180 #  initial stream is not valid XML,
181 # in this case we print directly to data
182 #  FIXME: we should test directly lxml.etree.XMLSyntaxError
183 # but importing lxml directly here is not clean
184 # should be wrapped in a custom Exception
185 self.disp(xml_data)
186 self.disp("")
187
188 async def start(self):
189 self.host.bridge.register_signal("xml_log", self.print_xml, "plugin")
190
191
192 class Theme(base.CommandBase):
193 def __init__(self, host):
194 base.CommandBase.__init__(
195 self, host, "theme", help=_("print colours used with your background")
196 )
197
198 def add_parser_options(self):
199 pass
200
201 async def start(self):
202 print(f"background currently used: {A.BOLD}{self.host.background}{A.RESET}\n")
203 for attr in dir(C):
204 if not attr.startswith("A_"):
205 continue
206 color = getattr(C, attr)
207 if attr == "A_LEVEL_COLORS":
208 # This constant contains multiple colors
209 self.disp("LEVEL COLORS: ", end=" ")
210 for idx, c in enumerate(color):
211 last = idx == len(color) - 1
212 end = "\n" if last else " "
213 self.disp(
214 c + f"LEVEL_{idx}" + A.RESET + (", " if not last else ""), end=end
215 )
216 else:
217 text = attr[2:]
218 self.disp(A.color(color, text))
219 self.host.quit()
220
221
222 class Debug(base.CommandBase):
223 subcommands = (bridge, Monitor, Theme)
224
225 def __init__(self, host):
226 super(Debug, self).__init__(
227 host, "debug", use_profile=False, help=_("debugging tools")
228 )