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

   1 import os.path as path
   2 import ast, json
   3
   4 from .lcnodes import LCModuleNode
   5 from .api import (
   6     ConfigLoadException,
   7     Renderable
   8 )
   9
  10 class LConfigFile:
  11     def __init__(self,
  12                  env,
  13                  file) -> None:
  14         self.__env = env
  15         self.__file = env.to_wspath(file)
  16
  17         with open(file) as f:
  18             tree = ast.parse(f.read(), self.__file, mode='exec')
  19             self.__module = LCModuleNode(self, tree)
  20
  21     def filename(self):
  22         return self.__file
  23
  24     def env(self):
  25         return self.__env
  26
  27     def root_module(self):
  28         return self.__module
  29
  30     def compile_astns(self, astns):
  31         if not isinstance(astns, list):
  32             astns = [astns]
  33
  34         return compile(
  35             ast.Module(body=astns, type_ignores=[]),
  36             self.__file, mode='exec')
  37
  38 class DependencyGraph:
  39     def __init__(self) -> None:
  40         self._edges = {}
  41
  42     def add(self, dependent, dependee):
  43         if dependent in self._edges:
  44             if dependee in self._edges[dependent]:
  45                 return
  46             self._edges[dependent].add(dependee)
  47         else:
  48             self._edges[dependent] = set([dependee])
  49
  50         if self.__check_loop(dependee):
  51             raise ConfigLoadException(
  52                 f"loop dependency detected: {dependent.get_name()}",
  53                 dependee)
  54
  55     def __check_loop(self, start):
  56         visited = set()
  57         q = [start]
  58         while len(q) > 0:
  59             current = q.pop()
  60             if current in visited:
  61                 return True
  62
  63             visited.add(current)
  64             if current not in self._edges:
  65                 continue
  66             for x in self._edges[current]:
  67                 q.append(x)
  68
  69         return False
  70
  71     def cascade(self, start):
  72         q = [start]
  73         while len(q) > 0:
  74             current = q.pop()
  75             if current in self._edges:
  76                 for x in self._edges[current]:
  77                     q.append(x)
  78             current.evaluate()
  79
  80 class ConfigTypeFactory:
  81     def __init__(self) -> None:
  82         self.__providers = []
  83
  84     def regitser(self, provider_type):
  85         self.__providers.append(provider_type)
  86
  87     def create(self, typedef):
  88         for provider in self.__providers:
  89             if not provider.typedef_matched(typedef):
  90                 continue
  91             return provider(typedef)
  92
  93         raise ConfigLoadException(
  94                 f"no type provider defined for type: {typedef}", None)
  95
  96 class LConfigEvaluationWrapper:
  97     def __init__(self, env, node) -> None:
  98         self.__env = env
  99         self.__node = node
 100
 101     def __enter__(self):
 102         self.__env.push_executing_node(self.__node)
 103         return self
 104
 105     def __exit__(self, type, value, tb):
 106         self.__env.pop_executing_node()
 107
 108     def evaluate(self):
 109         return self.__env.evaluate_node(self.__node)
 110
 111 class LConfigEnvironment(Renderable):
 112     def __init__(self, workspace, config_io) -> None:
 113         super().__init__()
 114
 115         self.__ws_path = path.abspath(workspace)
 116         self.__exec_globl = globals()
 117         self.__eval_stack = []
 118         self.__lc_modules = []
 119         self.__config_val = {}
 120         self.__node_table = {}
 121         self.__deps_graph = DependencyGraph()
 122         self.__type_fatry = ConfigTypeFactory()
 123         self.__config_io  = config_io
 124
 125     def to_wspath(self, _path):
 126         _path = path.abspath(_path)
 127         return path.relpath(_path, self.__ws_path)
 128
 129     def register_builtin_func(self, func_obj):
 130         call = (lambda *arg, **kwargs:
 131                      func_obj(self, *arg, **kwargs))
 132
 133         self.__exec_globl[func_obj.name] = call
 134
 135     def resolve_module(self, file):
 136         fo = LConfigFile(self, file)
 137         self.__lc_modules.insert(0, (fo.root_module()))
 138
 139     def evaluate_node(self, node):
 140         name = node.get_name()
 141
 142         return self.get_symbol(name)()
 143
 144     def eval_context(self, node):
 145         return LConfigEvaluationWrapper(self, node)
 146
 147     def push_executing_node(self, node):
 148         self.__eval_stack.append(node)
 149
 150     def pop_executing_node(self):
 151         return self.__eval_stack.pop()
 152
 153     def register_node(self, node):
 154         self.__node_table[node.get_name()] = node
 155
 156         l = {}
 157         try:
 158             self.push_executing_node(node)
 159             exec(node.get_co(), self.__exec_globl, l)
 160             self.pop_executing_node()
 161         except Exception as e:
 162             raise ConfigLoadException("failed to load", node, e)
 163
 164         self.__exec_globl.update(l)
 165
 166     def lookup_node(self, name):
 167         if name not in self.__node_table:
 168             raise ConfigLoadException(f"node '{name}' undefined")
 169         return self.__node_table[name]
 170
 171     def type_factory(self):
 172         return self.__type_fatry
 173
 174     def update_value(self, key, value):
 175         self.__config_val[key] = value
 176
 177     def get_symbol(self, name):
 178         if name not in self.__exec_globl:
 179             raise ConfigLoadException(f"unknown symbol '{name}'")
 180         return self.__exec_globl[name]
 181
 182     def callframe_at(self, traverse_level):
 183         try:
 184             return self.__eval_stack[traverse_level - 1]
 185         except:
 186             return None
 187
 188     def lookup_value(self, key):
 189         return self.__config_val[key]
 190
 191     def dependency(self):
 192         return self.__deps_graph
 193
 194     def update(self):
 195         for mod in self.__lc_modules:
 196             mod.evaluate()
 197
 198     def export(self):
 199         self.__config_io.export(self, self.__config_val)
 200
 201     def save(self, _path = ".config.json"):
 202         data = {}
 203         for mod in self.__lc_modules:
 204             mod.serialise(data)
 205
 206         with open(_path, 'w') as f:
 207             json.dump(data, f)
 208
 209     def load(self, _path = ".config.json"):
 210         if not path.exists(_path):
 211             return
 212
 213         data = {}
 214         with open(_path, 'r') as f:
 215             data.update(json.load(f))
 216
 217         for mod in self.__lc_modules:
 218             mod.deserialise(data)
 219
 220         self.update()
 221
 222     def render(self, rctx):
 223         for mod in self.__lc_modules:
 224             mod.render(rctx)