lunaix-os/scripts/build-tools/integration/render_ishell.py

   1 from lcfg.api import (
   2     RenderContext,
   3     InteractiveRenderer,
   4     Renderable,
   5     ConfigTypeCheckError,
   6     ConfigLoadException
   7 )
   8 from lcfg.common import LConfigEnvironment
   9
  10 import shlex
  11 from lib.utils import join_path
  12 from pathlib import Path
  13 import readline
  14
  15 class ShellException(Exception):
  16     def __init__(self, *args: object) -> None:
  17         super().__init__(*args)
  18
  19 class ViewElement:
  20     def __init__(self, label, node) -> None:
  21         self.label = label
  22         self.node = node
  23
  24     def expand(self, sctx):
  25         return False
  26
  27     def read(self, sctx):
  28         return None
  29
  30     def write(self, sctx, val):
  31         return None
  32
  33     def get_help(self, sctx):
  34         return self.node.help_prompt()
  35
  36     def get_type(self, sctx):
  37         return "N/A"
  38
  39     def draw(self, sctx, canvas):
  40         pass
  41
  42
  43 class SubviewLink(ViewElement):
  44     def __init__(self, label, node, cb) -> None:
  45         super().__init__(label, node)
  46         self.__callback = cb
  47
  48     def expand(self, sctx):
  49         sctx.clear_view()
  50         self.__callback(sctx)
  51         return True
  52
  53     def draw(self, sctx, canvas):
  54         print(f" [{self.label}]")
  55
  56     def get_type(self, sctx):
  57         return f"Collection: {self.label}"
  58
  59 class RootWrapper(ViewElement):
  60     def __init__(self, root) -> None:
  61         self.__root = root
  62         super().__init__("root", None)
  63
  64     def expand(self, sctx):
  65         sctx.clear_view()
  66         self.__root.render(sctx)
  67         return True
  68
  69     def draw(self, sctx, canvas):
  70         pass
  71
  72     def get_type(self, sctx):
  73         return ""
  74
  75 class FieldView(ViewElement):
  76     def __init__(self, label, node) -> None:
  77         super().__init__(label, node)
  78
  79     def read(self, sctx):
  80         return self.node.get_value()
  81
  82     def write(self, sctx, val):
  83         if self.node.read_only():
  84             return None
  85
  86         self.node.set_value(val)
  87         return val
  88
  89     def get_type(self, sctx):
  90         return f"Config term: {self.label}\n{self.node.get_type()}"
  91
  92     def draw(self, sctx, canvas):
  93         suffix = ""
  94         if self.node.read_only():
  95             suffix = " (ro)"
  96         print(f" {self.label}{suffix}")
  97
  98 class ShellContext(RenderContext):
  99     def __init__(self) -> None:
 100         super().__init__()
 101
 102         self._view = {}
 103         self._subviews = []
 104         self._field = []
 105
 106     def add_expandable(self, label, node, on_expand_cb):
 107         name = node.get_name()
 108         self._view[name] = SubviewLink(name, node, on_expand_cb)
 109         self._subviews.append(name)
 110
 111     def add_field(self, label, node):
 112         name = node.get_name()
 113         self._view[name] = FieldView(name, node)
 114         self._field.append(name)
 115
 116     def clear_view(self):
 117         self._view.clear()
 118         self._subviews.clear()
 119         self._field.clear()
 120
 121     def redraw(self):
 122         for v in self._subviews + self._field:
 123             self._view[v].draw(self, None)
 124
 125     def get_view(self, label):
 126         if label in self._view:
 127             return self._view[label]
 128         return None
 129
 130 class InteractiveShell(InteractiveRenderer):
 131     def __init__(self, root: Renderable) -> None:
 132         super().__init__()
 133         self.__levels = [RootWrapper(root)]
 134         self.__aborted = True
 135         self.__sctx = ShellContext()
 136         self.__cmd = {
 137             "ls": (
 138                 "list all config node under current collection",
 139                 lambda *args:
 140                     self.__sctx.redraw()
 141             ),
 142             "help": (
 143                 "print help prompt for given name of node",
 144                 lambda *args:
 145                     self.print_help(*args)
 146             ),
 147             "type": (
 148                 "print the type of a config node",
 149                 lambda *args:
 150                     self.print_type(*args)
 151             ),
 152             "cd": (
 153                 "navigate to a collection node given a unix-like path",
 154                 lambda *args:
 155                     self.get_in(*args)
 156             ),
 157             "usage": (
 158                 "print out the usage",
 159                 lambda *args:
 160                     self.print_usage()
 161             )
 162         }
 163
 164     def get_path(self):
 165         l = [level.label for level in self.__levels[1:]]
 166         return f"/{'/'.join(l)}"
 167
 168     def resolve(self, relative):
 169         ctx = ShellContext()
 170         p = join_path(self.get_path(), relative)
 171         ps = Path(p).resolve().parts
 172         if ps[0] == '/':
 173             ps = ps[1:]
 174
 175         node = self.__levels[0]
 176         levels = [node]
 177         for part in ps:
 178             if not node.expand(ctx):
 179                 raise ShellException(f"node is not a collection: {part}")
 180
 181             node = ctx.get_view(part)
 182             if not node:
 183                 raise ShellException(f"no such node: {part}")
 184
 185             levels.append(node)
 186
 187         return (node, levels)
 188
 189     def print_usage(self):
 190         for cmd, (desc, _) in self.__cmd.items():
 191             print("\n".join([
 192                 cmd,
 193                 f"   {desc}",
 194                 ""
 195             ]))
 196
 197     def print_help(self, node_name):
 198         view, _ = self.resolve(node_name)
 199
 200         print(view.get_help(self.__sctx))
 201
 202     def print_type(self, node_name):
 203         view, _ = self.resolve(node_name)
 204
 205         print(view.get_type(self.__sctx))
 206
 207     def do_read(self, node_name):
 208         view, _ = self.resolve(node_name)
 209         rd_val = view.read(self.__sctx)
 210         if rd_val is None:
 211             raise ShellException(f"config node {view.label} is not readable")
 212
 213         print(rd_val)
 214
 215     def do_write(self, node_name, val):
 216         view, _ = self.resolve(node_name)
 217         wr = view.write(self.__sctx, val)
 218         if not wr:
 219             raise ShellException(f"config node {view.label} is read only")
 220
 221         print(f"write: {val}")
 222         self.do_render()
 223
 224     def get_in(self, node_name):
 225         view, lvl = self.resolve(node_name)
 226
 227         if not view.expand(self.__sctx):
 228             print(f"{node_name} is not a collection")
 229             return
 230
 231         self.__levels = lvl
 232
 233     def do_render(self):
 234
 235         curr = self.__levels[-1]
 236         curr.expand(self.__sctx)
 237
 238     def __loop(self):
 239         prefix = "config: %s> "%(self.__levels[-1].label)
 240
 241         line = input(prefix)
 242         args = shlex.split(line)
 243         if not args:
 244             return True
 245
 246         cmd = args[0]
 247         if cmd in self.__cmd:
 248             self.__cmd[cmd][1](*args[1:])
 249             return True
 250
 251         if cmd == "exit":
 252             self.__aborted = False
 253             return False
 254
 255         if cmd == "abort":
 256             return False
 257
 258         node = self.resolve(cmd)
 259         if not node:
 260             raise ShellException(f"unrecognised command {line}")
 261
 262         if len(args) == 3 and args[1] == '=':
 263             self.do_write(args[0], args[2])
 264             return True
 265
 266         if len(args) == 1:
 267             self.do_read(args[0])
 268             return True
 269
 270         print(f"unrecognised command {line}")
 271         return True
 272
 273
 274     def render_loop(self):
 275         self.do_render()
 276         print("\n".join([
 277             "Interactive LConfig Shell",
 278             "   type 'usage' to find out how to use",
 279             "   type 'exit' to end (with saving)",
 280             "   type 'abort' to abort (without saving)",
 281             "   type node name directly to read the value",
 282             "   type 'node = val' to set 'val' to 'node'",
 283             ""
 284         ]))
 285         while True:
 286             try:
 287                 if not self.__loop():
 288                     break
 289             except KeyboardInterrupt:
 290                 break
 291             except ConfigTypeCheckError as e:
 292                 print(e.args[0])
 293             except ShellException as e:
 294                 print(e.args[0])
 295
 296         return not self.__aborted