"""Hatchling build hook that generates the ``scrypted_sdk`` package. The modules in this directory are flat top-level modules (symlinks into ``../../server/python`` and ``../../sdk/types``) because that is how the plugin runtime loads them. Publishing them to PyPI as-is would install top-level modules named ``rpc``, ``plugin_remote``, etc. into consumers' environments, so at build time this hook: 1. copies each module (reading through the symlinks) into a generated ``scrypted_sdk/`` package directory, and 2. mechanically rewrites the flat imports to package-qualified ones, e.g. ``import rpc`` -> ``from scrypted_sdk import rpc``. Nothing else is modified, and the generated directory is never committed. Dynamic plugin-zip imports (``from scrypted_sdk import sdk_init2``, ``from main import ...`` inside loadZip) are deliberately not rewritten: they resolve inside a plugin zip at runtime and are never reached by client usage. """ import re import shutil from pathlib import Path from hatchling.builders.hooks.plugin.interface import BuildHookInterface PACKAGE = "scrypted_sdk" # flat module -> module name inside the package. Only what a client needs: # the rpc/engine.io machinery and the types. The plugin host modules # (cluster_labels, plugin_console, plugin_pip, plugin_repl, plugin_volume) # are imported lazily by plugin_remote inside plugin-host-only code paths # and are deliberately not packaged. MODULES = { "rpc": "rpc", "rpc_reader": "rpc_reader", "plugin_remote": "plugin_remote", "cluster_setup": "cluster_setup", "scrypted_client": "client", # scrypted_python is a directory tree, handled separately but # rewritten with the same rule "scrypted_python": "scrypted_python", } INIT_PY = '''"""Python SDK for Scrypted. Generated at build time from the sources in packages/python-client (see hatch_build.py). """ from scrypted_sdk.client import ( DEFAULT_CONNECT_TIMEOUT, DEFAULT_PLUGIN_ID, EioRpcTransport, ScryptedConnectionError, connect_scrypted_client, ) from scrypted_sdk.plugin_remote import DeviceManager, MediaManager, SystemManager from scrypted_sdk.scrypted_python.scrypted_sdk import ScryptedStatic __all__ = [ "DEFAULT_CONNECT_TIMEOUT", "DEFAULT_PLUGIN_ID", "DeviceManager", "EioRpcTransport", "MediaManager", "ScryptedConnectionError", "ScryptedStatic", "SystemManager", "connect_scrypted_client", ] ''' _ALTERNATION = "|".join(sorted(MODULES, key=len, reverse=True)) _FROM_RE = re.compile(rf"^(\s*)from ({_ALTERNATION})((?:\.[\w.]+)?) import ") _IMPORT_RE = re.compile(rf"^(\s*)import ({_ALTERNATION})((?:\.[\w.]+)?)(\s+as\s+\w+)?\s*$") def _rewrite_line(line: str) -> str: m = _FROM_RE.match(line) if m: indent, mod, dots = m.groups() rest = line[m.end():] return f"{indent}from {PACKAGE}.{MODULES[mod]}{dots} import {rest}" m = _IMPORT_RE.match(line) if m: indent, mod, dots, alias = m.groups() target = MODULES[mod] if not dots: suffix = alias or (f" as {mod}" if target != mod else "") return f"{indent}from {PACKAGE} import {target}{suffix}\n" if alias: return f"{indent}import {PACKAGE}.{target}{dots}{alias}\n" # a dotted import without an alias binds the top-level name and # loads the submodules; preserve both return ( f"{indent}from {PACKAGE} import {target}; " f"import {PACKAGE}.{target}{dots} # noqa: E702\n" ) return line def _generate_file(src: Path, dest: Path) -> None: header = f"# Generated by hatch_build.py from {src.name} — do not edit.\n" text = "".join(_rewrite_line(l) for l in src.read_text().splitlines(keepends=True)) dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(header + text) def generate(source_dir: Path, package_dir: Path) -> None: if package_dir.exists(): shutil.rmtree(package_dir) package_dir.mkdir(parents=True) (package_dir / "__init__.py").write_text(INIT_PY) for flat, target in MODULES.items(): src = source_dir / f"{flat}.py" if src.exists(): _generate_file(src, package_dir / f"{target}.py") # the scrypted_python type tree (namespace package upstream; materialize # __init__.py so it is a regular subpackage) tree = source_dir / "scrypted_python" for src in sorted(tree.rglob("*.py")): if "__pycache__" in src.parts: continue _generate_file(src, package_dir / "scrypted_python" / src.relative_to(tree)) for d in [package_dir / "scrypted_python", *(p for p in (package_dir / "scrypted_python").rglob("*") if p.is_dir())]: init = d / "__init__.py" if not init.exists(): init.write_text("") class ScryptedSdkBuildHook(BuildHookInterface): def initialize(self, version: str, build_data: dict) -> None: root = Path(self.root) package_dir = root / ".build" / PACKAGE if (root / "plugin_remote.py").exists(): # building from the repo: generate the package generate(root, package_dir) # else: building a wheel from an sdist, which already ships the # generated package at .build/scrypted_sdk if self.target_name == "sdist": # keep it under .build/ in the sdist so hatchling's default # package detection doesn't also pick it up during the # wheel-from-sdist build build_data["force_include"][str(package_dir)] = f".build/{PACKAGE}" else: build_data["force_include"][str(package_dir)] = PACKAGE