lunaix-os/scripts/build-tools/lcfg/lcnodes.py

   1 from .api import (
   2     ConfigLoadException,
   3     ConfigTypeCheckError,
   4     Renderable
   5 )
   6
   7 from .utils import (
   8     extract_decorators,
   9     to_displayable
  10 )
  11
  12 import ast, textwrap
  13 from abc import abstractmethod
  14
  15
  16
  17 class LCNode(Renderable):
  18     def __init__(self, fo, astn):
  19         super().__init__()
  20
  21         self._fo = fo
  22         self._env = fo.env()
  23         self._astn = self.prune_astn(astn)
  24         self._co = self.compile_astn()
  25         self._parent = None
  26
  27         self._env.register_node(self)
  28
  29     def prune_astn(self, astn):
  30         return astn
  31
  32     def compile_astn(self):
  33         return self._fo.compile_astns(self._astn)
  34
  35     def get_co(self):
  36         return self._co
  37
  38     def get_fo(self):
  39         return self._fo
  40
  41     def set_parent(self, new_parent):
  42         self._parent = new_parent
  43
  44     def parent(self):
  45         return self._parent
  46
  47     def add_child(self, node):
  48         pass
  49
  50     def remove_child(self, node):
  51         pass
  52
  53     def get_name(self):
  54         return f"LCNode: {self.__hash__()}"
  55
  56     @abstractmethod
  57     def evaluate(self):
  58         pass
  59
  60     def render(self, rctx):
  61         pass
  62
  63     def deserialise(self, dict):
  64         pass
  65
  66     def serialise(self, dict):
  67         pass
  68
  69
  70
  71 class LCModuleNode(LCNode):
  72     def __init__(self, fo, astn: ast.Module):
  73         self.__nodes = {}
  74
  75         super().__init__(fo, astn)
  76
  77     def get_name(self):
  78         return f"file: {self._fo.filename()}"
  79
  80     def prune_astn(self, astn: ast.Module):
  81         general_exprs = []
  82
  83         for b in astn.body:
  84             if not isinstance(b, ast.FunctionDef):
  85                 general_exprs.append(b)
  86                 continue
  87
  88             node = LCFuncNode.construct(self._fo, b)
  89             node.set_parent(self)
  90
  91             self.add_child(node)
  92
  93         return general_exprs
  94
  95     def evaluate(self):
  96         with self._env.eval_context(self) as evalc:
  97             ls = list(self.__nodes.values())
  98             for node in ls:
  99                 node.evaluate()
 100
 101     def add_child(self, node):
 102         self.__nodes[node] = node
 103
 104     def remove_child(self, node):
 105         if node in self.__nodes:
 106             del self.__nodes[node]
 107
 108     def deserialise(self, dict):
 109         for node in self.__nodes:
 110             node.deserialise(dict)
 111
 112     def serialise(self, dict):
 113         for node in self.__nodes:
 114             node.serialise(dict)
 115
 116     def render(self, rctx):
 117         for node in self.__nodes:
 118             node.render(rctx)
 119
 120 class LCFuncNode(LCNode):
 121     def __init__(self, fo, astn) -> None:
 122         self._decors = {}
 123         self._name = None
 124         self._help = ""
 125         self._display_name = None
 126         self._enabled = True
 127
 128         super().__init__(fo, astn)
 129
 130     def prune_astn(self, astn: ast.FunctionDef):
 131         self._name = astn.name
 132         self._display_name = to_displayable(self._name)
 133
 134         maybe_doc = astn.body[0]
 135         if isinstance(maybe_doc, ast.Expr):
 136             if isinstance(maybe_doc.value, ast.Constant):
 137                 self._help = maybe_doc.value.value
 138                 self._help = textwrap.dedent(self._help)
 139
 140         decors = extract_decorators(astn)
 141         for name, args, kwargs in decors:
 142             self._decors[name] = (args, kwargs)
 143
 144         (args, _) = self._decors[self.mapped_name()]
 145         if args:
 146             self._display_name = args[0]
 147
 148         astn.decorator_list.clear()
 149         return astn
 150
 151     def get_name(self):
 152         return self._name
 153
 154     def get_display_name(self):
 155         return self._display_name
 156
 157     def enabled(self):
 158         if isinstance(self._parent, LCFuncNode):
 159             return self._enabled and self._parent.enabled()
 160         return self._enabled
 161
 162     def help_prompt(self):
 163         return self._help
 164
 165     def evaluate(self):
 166         with self._env.eval_context(self) as evalc:
 167             result = evalc.evaluate()
 168             self._enabled = True if result is None else result
 169
 170     @staticmethod
 171     def mapped_name(self):
 172         return None
 173
 174     @staticmethod
 175     def construct(fo, astn: ast.FunctionDef):
 176         nodes = [
 177             LCCollectionNode,
 178             LCGroupNode,
 179             LCTermNode
 180         ]
 181
 182         for node in nodes:
 183             if extract_decorators(astn, node.mapped_name(), True):
 184                 return node(fo, astn)
 185
 186         raise ConfigLoadException(
 187                 f"unknown type for astn type: {type(astn)}")
 188
 189     def set_parent(self, new_parent):
 190         if self._parent:
 191             self._parent.remove_child(self)
 192
 193         new_parent.add_child(self)
 194         super().set_parent(new_parent)
 195
 196
 197 class LCTermNode(LCFuncNode):
 198     def __init__(self, fo, astn) -> None:
 199         self._value = None
 200         self._default = None
 201         self._type = None
 202         self._rdonly = False
 203         self._ready = False
 204
 205         super().__init__(fo, astn)
 206
 207     @staticmethod
 208     def mapped_name():
 209         return "Term"
 210
 211     def prune_astn(self, astn: ast.FunctionDef):
 212         astn = super().prune_astn(astn)
 213
 214         self._rdonly = "ReadOnly" in self._decors
 215
 216         return astn
 217
 218     def set_type(self, type_def):
 219         self._type = self._env.type_factory().create(type_def)
 220
 221     def get_type(self):
 222         return self._type
 223
 224     def __assert_type(self, val):
 225         if not self._type:
 226             raise ConfigLoadException(
 227                     f"config: {self._name} must be typed", self)
 228
 229         if self._type.check(val):
 230            return
 231
 232         raise ConfigTypeCheckError(
 233                 f"value: {val} does not match type {self._type}", self)
 234
 235     def set_value(self, val):
 236         if self._rdonly:
 237             return
 238
 239         if isinstance(val, str):
 240             val = self._type.parse_input(val)
 241
 242         self.__assert_type(val)
 243         self._value = val
 244
 245         self._ready = True
 246         self.__update_value()
 247         self._env.dependency().cascade(self)
 248
 249
 250     def set_default(self, val):
 251         self.__assert_type(val)
 252         self._default = val
 253
 254         if self._rdonly:
 255             self._value = val
 256
 257     def get_value(self):
 258         return self._value
 259
 260     def is_ready(self):
 261         return self._ready
 262
 263     def evaluate(self):
 264         super().evaluate()
 265         self.__update_value()
 266
 267     def read_only(self):
 268         return self._rdonly
 269
 270     def render(self, rctx):
 271         if self.enabled():
 272             rctx.add_field(self._display_name, self)
 273
 274     def serialise(self, dict):
 275         if not self.enabled():
 276             return
 277
 278         s_val = self._type.serialise(self._value)
 279         dict[self._name] = s_val
 280
 281     def deserialise(self, dict):
 282         if self._name not in dict or not self.enabled():
 283             return
 284
 285         s_val = dict[self._name]
 286         v = self._type.deserialise(s_val)
 287         self.__assert_type(v)
 288         self._value = v
 289
 290
 291     def __update_value(self):
 292         v = self._value
 293
 294         if not self.enabled():
 295             self._env.update_value(self._name, None)
 296             return
 297
 298         if v is None:
 299             v = self._default
 300
 301         if v is None and not self._type.allow_none():
 302             raise ConfigLoadException(
 303                     f"config: {self._name} must have a value", self)
 304
 305         self._value = v
 306         self._env.update_value(self._name, v)
 307
 308
 309
 310
 311 class LCGroupNode(LCFuncNode):
 312     def __init__(self, fo, astn) -> None:
 313         self._children = {}
 314
 315         super().__init__(fo, astn)
 316
 317     @staticmethod
 318     def mapped_name():
 319         return "Group"
 320
 321     def prune_astn(self, astn: ast.FunctionDef):
 322         astn = super().prune_astn(astn)
 323
 324         other_exprs = []
 325         for expr in astn.body:
 326             if not isinstance(expr, ast.FunctionDef):
 327                 other_exprs.append(expr)
 328                 continue
 329
 330             node = LCFuncNode.construct(self._fo, expr)
 331             node.set_parent(self)
 332             self._children[node] = node
 333
 334         if not other_exprs:
 335             other_exprs.append(
 336                 ast.Pass(
 337                     lineno=astn.lineno + 1,
 338                     col_offset=astn.col_offset
 339                 )
 340             )
 341
 342         astn.body = other_exprs
 343         return astn
 344
 345     def evaluate(self):
 346         old_enable = super().enabled()
 347         super().evaluate()
 348
 349         new_enabled = super().enabled()
 350         if not new_enabled and old_enable == new_enabled:
 351             return
 352
 353         with self._env.eval_context(self) as evalc:
 354             children = list(self._children.values())
 355             for child in children:
 356                 child.evaluate()
 357
 358     def render(self, rctx):
 359         if not self.enabled():
 360             return
 361
 362         for child in self._children.values():
 363             child.render(rctx)
 364
 365     def serialise(self, dict):
 366         sub_dict = {}
 367         for child in self._children.values():
 368             child.serialise(sub_dict)
 369
 370         dict[self._name] = sub_dict
 371
 372     def deserialise(self, dict):
 373         if self._name not in dict:
 374             return
 375
 376         sub_dict = dict[self._name]
 377         for child in self._children.values():
 378             child.deserialise(sub_dict)
 379
 380     def add_child(self, node):
 381         self._children[node] = node
 382
 383     def remove_child(self, node):
 384         if node in self._children:
 385             del self._children[node]
 386
 387
 388 class LCCollectionNode(LCGroupNode):
 389     def __init__(self, fo, astn) -> None:
 390         super().__init__(fo, astn)
 391
 392     @staticmethod
 393     def mapped_name():
 394         return "Collection"
 395
 396     def render(self, rctx):
 397         if not self.enabled():
 398             return
 399
 400         label = self._display_name
 401         rctx.add_expandable(label, self, self.__expand)
 402
 403     def __expand(self, _ctx):
 404         super().render(_ctx)