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         s_val = self._type.serialise(self._value)
 276         dict[self._name] = s_val
 277
 278     def deserialise(self, dict):
 279         if self._name not in dict:
 280             return
 281
 282         s_val = dict[self._name]
 283         v = self._type.deserialise(s_val)
 284         self.__assert_type(v)
 285         self._value = v
 286
 287
 288     def __update_value(self):
 289         v = self._value
 290
 291         if not self.enabled():
 292             self._env.update_value(self._name, None)
 293             return
 294
 295         if v is None:
 296             v = self._default
 297
 298         if v is None and not self._type.allow_none():
 299             raise ConfigLoadException(
 300                     f"config: {self._name} must have a value", self)
 301
 302         self._value = v
 303         self._env.update_value(self._name, v)
 304
 305
 306
 307
 308 class LCGroupNode(LCFuncNode):
 309     def __init__(self, fo, astn) -> None:
 310         self._children = {}
 311
 312         super().__init__(fo, astn)
 313
 314     @staticmethod
 315     def mapped_name():
 316         return "Group"
 317
 318     def prune_astn(self, astn: ast.FunctionDef):
 319         astn = super().prune_astn(astn)
 320
 321         other_exprs = []
 322         for expr in astn.body:
 323             if not isinstance(expr, ast.FunctionDef):
 324                 other_exprs.append(expr)
 325                 continue
 326
 327             node = LCFuncNode.construct(self._fo, expr)
 328             node.set_parent(self)
 329             self._children[node] = node
 330
 331         if not other_exprs:
 332             other_exprs.append(
 333                 ast.Pass(
 334                     lineno=astn.lineno + 1,
 335                     col_offset=astn.col_offset
 336                 )
 337             )
 338
 339         astn.body = other_exprs
 340         return astn
 341
 342     def evaluate(self):
 343         old_enable = super().enabled()
 344         super().evaluate()
 345
 346         new_enabled = super().enabled()
 347         if not new_enabled and old_enable == new_enabled:
 348             return
 349
 350         with self._env.eval_context(self) as evalc:
 351             children = list(self._children.values())
 352             for child in children:
 353                 child.evaluate()
 354
 355     def render(self, rctx):
 356         for child in self._children.values():
 357             child.render(rctx)
 358
 359     def serialise(self, dict):
 360         sub_dict = {}
 361         for child in self._children.values():
 362             child.serialise(sub_dict)
 363
 364         dict[self._name] = sub_dict
 365
 366     def deserialise(self, dict):
 367         if self._name not in dict:
 368             return
 369
 370         sub_dict = dict[self._name]
 371         for child in self._children.values():
 372             child.deserialise(sub_dict)
 373
 374     def add_child(self, node):
 375         self._children[node] = node
 376
 377     def remove_child(self, node):
 378         if node in self._children:
 379             del self._children[node]
 380
 381
 382 class LCCollectionNode(LCGroupNode):
 383     def __init__(self, fo, astn) -> None:
 384         super().__init__(fo, astn)
 385
 386     @staticmethod
 387     def mapped_name():
 388         return "Collection"
 389
 390     def render(self, rctx):
 391         _super = super()
 392         rctx.add_expandable(
 393             self._display_name,
 394             self,
 395             lambda _ctx:
 396                 _super.render(_ctx)
 397         )