scrypted/packages/python-client/hatch_build.py

"""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,
    PluginRemote,
    SystemManager,
)
from scrypted_sdk.scrypted_python.scrypted_sdk import ScryptedStatic
from scrypted_sdk.scrypted_python.scrypted_sdk.types import (
    ScryptedDeviceType,
    ScryptedInterface,
    ScryptedInterfaceMethods,
    ScryptedInterfaceProperty,
    ScryptedMimeTypes,
)

__all__ = [
    "DEFAULT_CONNECT_TIMEOUT",
    "DEFAULT_PLUGIN_ID",
    "DeviceManager",
    "EioRpcTransport",
    "MediaManager",
    "PluginRemote",
    "ScryptedConnectionError",
    "ScryptedDeviceType",
    "ScryptedInterface",
    "ScryptedInterfaceMethods",
    "ScryptedInterfaceProperty",
    "ScryptedMimeTypes",
    "ScryptedStatic",
    "SystemManager",
    "connect_scrypted_client",
]
'''

TYPES_PY = '''"""Alias for the full generated type surface.

The concrete type tree lives at scrypted_sdk.scrypted_python.scrypted_sdk.types
(an artifact of build-time package generation); this module exposes all of it
in one place::

    from scrypted_sdk.types import ScryptedInterface, VideoClip, LockState

The package root re-exports only the curated common names; everything else
(interface TypedDicts, settings types, the remaining enums) is available here.
"""

from scrypted_sdk.scrypted_python.scrypted_sdk.types import *  # noqa: F401,F403
'''

_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)
    (package_dir / "types.py").write_text(TYPES_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