96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
"""Reviewable import closure for the onboard plugin; no entire Core tree."""
|
|
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
MANIFEST = Path(__file__).with_name("runtime-files.json")
|
|
ENTRYPOINT = "k1link.device_plugins.xgrids_k1.node_bridge"
|
|
|
|
|
|
def runtime_nodes(node):
|
|
# A lazy package's typing-only reexports must not ship every LAB algorithm.
|
|
if isinstance(node, ast.If) and (
|
|
isinstance(node.test, ast.Name)
|
|
and node.test.id == "TYPE_CHECKING"
|
|
or isinstance(node.test, ast.Attribute)
|
|
and node.test.attr == "TYPE_CHECKING"
|
|
):
|
|
for child in node.orelse:
|
|
yield from runtime_nodes(child)
|
|
return
|
|
yield node
|
|
for child in ast.iter_child_nodes(node):
|
|
yield from runtime_nodes(child)
|
|
|
|
|
|
def import_closure(repository=ROOT):
|
|
source = repository / "src"
|
|
modules = {}
|
|
for path in (source / "k1link").rglob("*.py"):
|
|
name = ".".join(path.relative_to(source).with_suffix("").parts).removesuffix(".__init__")
|
|
modules[name] = path
|
|
seen, todo = set(), [ENTRYPOINT]
|
|
while todo:
|
|
name = todo.pop()
|
|
if name in seen:
|
|
continue
|
|
seen.add(name)
|
|
path = modules[name]
|
|
package = name if path.name == "__init__.py" else name.rpartition(".")[0]
|
|
imports = {".".join(name.split(".")[:i]) for i in range(1, len(name.split(".")))}
|
|
for node in runtime_nodes(ast.parse(path.read_text())):
|
|
if isinstance(node, ast.Import):
|
|
imports.update(a.name for a in node.names)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
base = node.module or ""
|
|
if node.level:
|
|
prefix = ".".join(
|
|
package.split(".")[: len(package.split(".")) - node.level + 1]
|
|
)
|
|
base = prefix + ("." + base if base else "")
|
|
imports.add(base)
|
|
imports.update(base + "." + a.name for a in node.names)
|
|
todo.extend(sorted((imports & modules.keys()) - seen))
|
|
return sorted(str(modules[name].relative_to(repository)) for name in seen)
|
|
|
|
|
|
def files(repository=ROOT):
|
|
declared = json.loads(MANIFEST.read_text())
|
|
actual = import_closure(repository)
|
|
if (
|
|
declared["schema"] != "missioncore.plugin.python-payload/v1"
|
|
or declared["entrypoint"] != ENTRYPOINT
|
|
or actual != declared["files"]
|
|
):
|
|
raise ValueError("K1 runtime imports changed; review its payload manifest before packaging")
|
|
result = []
|
|
for name in actual:
|
|
path = repository / name
|
|
if path.is_symlink():
|
|
raise ValueError("Symlink in plugin source payload")
|
|
result.append(path)
|
|
return result
|
|
|
|
|
|
def provenance(repository=ROOT):
|
|
return {
|
|
str(p.relative_to(repository)): hashlib.sha256(p.read_bytes()).hexdigest()
|
|
for p in files(repository)
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"schema": "missioncore.plugin.python-payload/v1",
|
|
"entrypoint": ENTRYPOINT,
|
|
"files": import_closure(),
|
|
},
|
|
indent=2,
|
|
)
|
|
)
|