Merge branch 'main' of github.com:koush/scrypted

.github/workflows/python-sdk.yml

@@ -0,0 +1,48 @@
+name: Python SDK
+
+on:
+  push:
+    tags: ["python-sdk-v*"]
+  pull_request:
+    paths:
+      - "packages/python-client/**"
+      - "server/python/**"
+      - "sdk/types/scrypted_python/**"
+      - ".github/workflows/python-sdk.yml"
+
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: packages/python-client
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v5
+      - name: Build sdist and wheel
+        run: uv build
+      - name: Smoke test the wheel
+        run: |
+          uv venv /tmp/smoke
+          uv pip install --python /tmp/smoke/bin/python dist/*.whl
+          /tmp/smoke/bin/python -c "import scrypted_sdk; print(scrypted_sdk.connect_scrypted_client)"
+      - uses: actions/upload-artifact@v4
+        with:
+          name: python-sdk-dist
+          path: packages/python-client/dist/
+
+  publish:
+    name: Publish to PyPI
+    if: startsWith(github.ref, 'refs/tags/python-sdk-v')
+    needs: build
+    runs-on: ubuntu-latest
+    environment: pypi
+    permissions:
+      id-token: write # PyPI trusted publishing
+    steps:
+      - uses: actions/download-artifact@v4
+        with:
+          name: python-sdk-dist
+          path: dist/
+      - uses: pypa/gh-action-pypi-publish@release/v1

packages/python-client/.gitignore

@@ -1 +1,4 @@
 .venv
+.build
+dist
+__pycache__

packages/python-client/README.md

@@ -0,0 +1,96 @@
+# python-client
+
+Connect to a Scrypted server from Python and use the same SDK objects
+(`systemManager`, `deviceManager`, `mediaManager`) that Python plugins see.
+
+The modules in this directory are symlinks into `../../server/python` and
+`../../sdk/types` so the client and the plugin runtime share one
+implementation.
+
+## Usage
+
+Installed from PyPI, everything lives under the `scrypted_sdk` package:
+
+```bash
+pip install scrypted-sdk
+```
+
+```python
+from scrypted_sdk import connect_scrypted_client
+```
+
+From a checkout of this repository, the modules are importable directly:
+
+```bash
+pip install -r requirements.txt
+```
+
+```python
+import asyncio
+
+from scrypted_client import connect_scrypted_client
+
+
+async def main(loop):
+    transport, sdk = await connect_scrypted_client(
+        loop, "https://localhost:10443", "username", "password"
+    )
+    for id in sdk.systemManager.getSystemState():
+        print(sdk.systemManager.getDeviceById(id).name)
+    await transport.close()
+
+
+loop = asyncio.new_event_loop()
+loop.run_until_complete(main(loop))
+```
+
+`test.py` is a runnable version of the above; point it at a server with the
+`SCRYPTED_BASE_URL`, `SCRYPTED_USERNAME`, and `SCRYPTED_PASSWORD` environment
+variables.
+
+[examples/light.py](examples/light.py) is the Python equivalent of
+[packages/client/examples/light.ts](../client/examples/light.ts) — it turns a
+named light on and back off:
+
+```bash
+SCRYPTED_USERNAME=admin SCRYPTED_PASSWORD=swordfish python examples/light.py "Office Dimmer"
+```
+
+It works both with `pip install scrypted-sdk` and directly from a repo
+checkout with only `requirements.txt` installed.
+
+By default the login request and the engine.io connection skip TLS
+certificate verification, since Scrypted servers use self-signed certificates
+out of the box. To control TLS (or connection pooling), pass your own
+`login_session` and a pre-built `EioRpcTransport(loop, http_session=...)`.
+
+Session ownership: a `login_session` you pass is only borrowed for the login
+request and is never closed. An `http_session` passed to `EioRpcTransport`
+becomes owned by the transport — `transport.close()` closes it along with the
+engine.io connection and background tasks — so don't share that session with
+anything else.
+
+## Packaging
+
+The [scrypted-sdk](https://pypi.org/project/scrypted-sdk/) PyPI package is
+built from this directory. Because the modules here are flat top-level
+modules (that is how the plugin runtime loads them), publishing them as-is
+would install modules named `rpc`, `plugin_remote`, etc. into consumers'
+environments. Instead, a build hook ([hatch_build.py](hatch_build.py))
+generates a `scrypted_sdk` package at build time: it copies each module
+(reading through the symlinks) and mechanically rewrites the flat imports to
+package-qualified ones (`import rpc` → `from scrypted_sdk import rpc`).
+Nothing is committed and no runtime code is modified beyond that rewrite.
+
+To build locally:
+
+```bash
+pip install build && python -m build   # or: uv build
+```
+
+To release: bump `version` in [pyproject.toml](pyproject.toml) and push a
+`python-sdk-v*` tag. The [Python SDK workflow](../../.github/workflows/python-sdk.yml)
+builds, smoke-tests, and publishes to PyPI via
+[trusted publishing](https://docs.pypi.org/trusted-publishers/) — the PyPI
+project just needs `koush/scrypted` + `python-sdk.yml` registered as a
+trusted publisher (no API tokens).

packages/python-client/cluster_setup.py

@@ -0,0 +1 @@
+../../server/python/cluster_setup.py

packages/python-client/examples/light.py

@@ -0,0 +1,50 @@
+"""Turn a light on and off from the command line.
+
+The Python equivalent of packages/client/examples/light.ts.
+
+Usage:
+
+    pip install scrypted-sdk  # or, from a repo checkout: pip install packages/python-client
+    SCRYPTED_USERNAME=admin SCRYPTED_PASSWORD=swordfish python light.py "Office Dimmer"
+
+The server URL defaults to https://localhost:10443 and can be overridden
+with SCRYPTED_BASE_URL.
+"""
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+try:
+    from scrypted_sdk import connect_scrypted_client
+except ImportError:
+    # running from a repo checkout without the scrypted-sdk package
+    # installed: use the flat modules in the parent directory
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+    from scrypted_client import connect_scrypted_client
+
+
+async def example(loop: asyncio.AbstractEventLoop):
+    transport, sdk = await connect_scrypted_client(
+        loop,
+        os.environ.get("SCRYPTED_BASE_URL", "https://localhost:10443"),
+        os.environ.get("SCRYPTED_USERNAME", "admin"),
+        os.environ.get("SCRYPTED_PASSWORD", "swordfish"),
+        plugin_id="@scrypted/core",
+    )
+    print("connected,", len(sdk.systemManager.getSystemState()), "devices")
+
+    name = sys.argv[1] if len(sys.argv) > 1 else "Office Dimmer"
+    dimmer = sdk.systemManager.getDeviceByName(name)
+    if not dimmer:
+        raise Exception("Device not found")
+    await dimmer.turnOn()
+    await asyncio.sleep(5)
+    await dimmer.turnOff()
+    # allow python to exit
+    await transport.close()
+
+
+loop = asyncio.new_event_loop()
+loop.run_until_complete(example(loop))

packages/python-client/hatch_build.py

@@ -0,0 +1,150 @@
+"""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

packages/python-client/pyproject.toml

@@ -0,0 +1,40 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "scrypted-sdk"
+version = "0.1.0"
+description = "Python SDK for Scrypted: connect to a Scrypted server and use the same SDK objects plugins see"
+readme = "README.md"
+license = "ISC"
+requires-python = ">=3.10"
+authors = [
+    { name = "Koushik Dutta", email = "koushd@gmail.com" },
+]
+keywords = ["scrypted", "home-automation", "camera", "nvr"]
+classifiers = [
+    "Framework :: AsyncIO",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3",
+    "Topic :: Home Automation",
+]
+# keep in sync with requirements.txt
+dependencies = [
+    "aiohttp",
+    "aiodns",
+    "python-engineio[asyncio_client]",
+]
+
+[project.urls]
+Homepage = "https://github.com/koush/scrypted"
+Source = "https://github.com/koush/scrypted/tree/main/packages/python-client"
+
+[tool.hatch.build.targets.sdist]
+include = ["pyproject.toml", "hatch_build.py", "README.md"]
+
+[tool.hatch.build.targets.sdist.hooks.custom]
+path = "hatch_build.py"
+
+[tool.hatch.build.targets.wheel.hooks.custom]
+path = "hatch_build.py"

packages/python-client/scrypted_client.py

@@ -0,0 +1,249 @@
+"""Client library for connecting to a Scrypted server over engine.io RPC.
+
+Provides :class:`EioRpcTransport` and :func:`connect_scrypted_client`, the
+importable equivalent of the bootstrap that previously only existed inline in
+test.py. Typical usage::
+
+    transport, sdk = await connect_scrypted_client(
+        asyncio.get_event_loop(), "https://localhost:10443", username, password
+    )
+    for id in sdk.systemManager.getSystemState():
+        print(sdk.systemManager.getDeviceById(id).name)
+    await transport.close()
+
+Must be called from a running event loop; the transport schedules its send
+loop on construction.
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+import aiohttp
+import engineio
+
+import plugin_remote
+import rpc_reader
+from cluster_setup import ClusterSetup
+from plugin_remote import DeviceManager, MediaManager, SystemManager
+from scrypted_python.scrypted_sdk import ScryptedStatic
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_PLUGIN_ID = "@scrypted/core"
+DEFAULT_CONNECT_TIMEOUT = 30
+
+
+class ScryptedConnectionError(Exception):
+    """Raised when the Scrypted engine.io connection cannot be established."""
+
+
+class EioRpcTransport(rpc_reader.RpcTransport):
+    """RpcTransport over an engine.io connection."""
+
+    def __init__(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        http_session: aiohttp.ClientSession | None = None,
+    ) -> None:
+        super().__init__()
+        # When a session is provided, engineio must defer to its connector
+        # (ssl_verify=True skips engineio's own ssl.create_default_context,
+        # a blocking call some frameworks forbid on the event loop). Only
+        # fall back to engineio's ssl_verify=False when we have no session;
+        # Scrypted servers use self-signed certificates by default. engineio
+        # never closes externally provided sessions, so the transport takes
+        # ownership: close() closes the session. Callers that need to keep
+        # the session alive should not share it with the transport.
+        self._http_session = http_session
+        self.eio = engineio.AsyncClient(
+            http_session=http_session, ssl_verify=http_session is not None
+        )
+        self.loop = loop
+        self.write_error: Exception | None = None
+        self.read_queue: asyncio.Queue = asyncio.Queue()
+        self.write_queue: asyncio.Queue = asyncio.Queue()
+        self._send_task: asyncio.Task | None = None
+        # Set by connect_scrypted_client() so close() can tear it down.
+        self.read_task: asyncio.Task | None = None
+
+        @self.eio.on("message")
+        def on_message(data):
+            self.read_queue.put_nowait(data)
+
+        self._send_task = loop.create_task(self.send_loop())
+
+    async def read(self):
+        return await self.read_queue.get()
+
+    async def send_loop(self):
+        while True:
+            data = await self.write_queue.get()
+            try:
+                await self.eio.send(data)
+            except Exception as e:
+                # Any send failure kills the link; surface it on next write.
+                self.write_error = e
+                break
+
+    def writeBuffer(self, buffer, reject):
+        if self.write_error:
+            if reject:
+                reject(self.write_error)
+            return
+        self.write_queue.put_nowait(buffer)
+
+    def writeSerialized(self, j, reject):
+        # engineio json-encodes dict payloads on send and json-decodes text
+        # frames on receive, so messages cross the wire as engine.io JSON
+        # packets and arrive at readLoop() already deserialized to dicts.
+        return self.writeBuffer(j, reject)
+
+    async def close(self) -> None:
+        """Tear the transport down, cancelling background tasks."""
+        tasks = [task for task in (self._send_task, self.read_task) if task]
+        self._send_task = None
+        self.read_task = None
+        for task in tasks:
+            task.cancel()
+        try:
+            await self.eio.disconnect()
+        except Exception:
+            logger.debug("Error disconnecting engine.io client", exc_info=True)
+        for task in tasks:
+            with contextlib.suppress(asyncio.CancelledError, Exception):
+                await task
+        if self._http_session:
+            await self._http_session.close()
+            self._http_session = None
+
+
+async def connect_scrypted_client(
+    loop: asyncio.AbstractEventLoop,
+    base_url: str,
+    username: str,
+    password: str,
+    plugin_id: str = DEFAULT_PLUGIN_ID,
+    login_session: aiohttp.ClientSession | None = None,
+    transport: EioRpcTransport | None = None,
+    timeout: float = DEFAULT_CONNECT_TIMEOUT,
+) -> tuple[EioRpcTransport, ScryptedStatic]:
+    """Login and establish the engine.io RPC session.
+
+    Returns ``(transport, sdk)``. The caller owns the transport and must
+    ``await transport.close()`` when done. If ``login_session`` is omitted, a
+    temporary session with certificate verification disabled is used for the
+    login request only; pass a configured session to control TLS behavior.
+    Likewise pass a pre-built :class:`EioRpcTransport` to control the
+    engine.io connection's session (the transport takes ownership of that
+    session and closes it in ``close()``). ``timeout`` bounds each phase of
+    the connection: the login request, the engine.io connect, and the wait
+    for initial system state. Raises :class:`ScryptedConnectionError` on any
+    failure.
+    """
+    owns_login_session = login_session is None
+    session = login_session or aiohttp.ClientSession(
+        connector=aiohttp.TCPConnector(ssl=False)
+    )
+    try:
+        async with session.post(
+            f"{base_url}/login",
+            json={"username": username, "password": password},
+            raise_for_status=True,
+            timeout=aiohttp.ClientTimeout(total=timeout),
+        ) as response:
+            login_response = await response.json()
+    except (aiohttp.ClientError, asyncio.TimeoutError) as err:
+        raise ScryptedConnectionError(f"Login to {base_url} failed: {err}") from err
+    finally:
+        if owns_login_session:
+            await session.close()
+
+    if "authorization" not in login_response:
+        raise ScryptedConnectionError(
+            f"Login to {base_url} did not return an authorization header "
+            f"(response keys: {sorted(login_response)})"
+        )
+
+    if transport is None:
+        transport = EioRpcTransport(loop)
+    try:
+        await asyncio.wait_for(
+            transport.eio.connect(
+                base_url,
+                headers={"Authorization": login_response["authorization"]},
+                engineio_path=f"/endpoint/{plugin_id}/engine.io/api/",
+            ),
+            timeout,
+        )
+    except (engineio.exceptions.ConnectionError, asyncio.TimeoutError) as err:
+        await transport.close()
+        raise ScryptedConnectionError(f"engine.io connect failed: {err}") from err
+
+    ret: asyncio.Future[ScryptedStatic] = loop.create_future()
+    peer, peer_read_loop = await rpc_reader.prepare_peer_readloop(loop, transport)
+    peer.params["print"] = logger.debug
+
+    def get_remote(api, plugin_id_, host_info):
+        cluster_setup = ClusterSetup(loop, peer)
+        remote = plugin_remote.PluginRemote(
+            cluster_setup, api, plugin_id_, host_info, loop
+        )
+        wrapped = remote.setSystemState
+
+        async def remote_set_system_state(system_state):
+            await wrapped(system_state)
+
+            async def resolve():
+                if ret.done():
+                    return
+                sdk = ScryptedStatic()
+                sdk.api = api
+                sdk.remote = remote
+                system_manager = SystemManager(api, remote.systemState)
+                sdk.systemManager = system_manager
+                sdk.deviceManager = DeviceManager(remote.nativeIds, system_manager)
+                sdk.mediaManager = MediaManager(await api.getMediaManager())
+                if not ret.done():
+                    # PluginRemote.notify dispatches events to the attached
+                    # systemManager's registry (same wiring as loadZip); this
+                    # is what makes systemManager.listen() callbacks fire.
+                    remote.systemManager = system_manager
+                    ret.set_result(sdk)
+
+            loop.create_task(resolve())
+
+        remote.setSystemState = remote_set_system_state
+        return remote
+
+    peer.params["getRemote"] = get_remote
+    read_task = loop.create_task(peer_read_loop())
+    transport.read_task = read_task
+
+    def on_read_loop_done(task: asyncio.Task) -> None:
+        # Surface a dead link immediately instead of waiting for the timeout.
+        if ret.done() or task.cancelled():
+            return
+        err = task.exception()
+        ret.set_exception(
+            ScryptedConnectionError(
+                f"Connection to {base_url} lost during handshake: {err}"
+                if err
+                else f"Connection to {base_url} closed during handshake"
+            )
+        )
+
+    read_task.add_done_callback(on_read_loop_done)
+
+    try:
+        sdk = await asyncio.wait_for(asyncio.shield(ret), timeout)
+    except asyncio.TimeoutError as err:
+        await transport.close()
+        raise ScryptedConnectionError(
+            f"Timed out waiting for Scrypted system state from {base_url}"
+        ) from err
+    except ScryptedConnectionError:
+        await transport.close()
+        raise
+    return transport, sdk

packages/python-client/test.py

@@ -2,136 +2,15 @@ from __future__ import annotations
 
 import asyncio
 import os
-from contextlib import nullcontext
 
-import aiohttp
-import engineio
-
-import plugin_remote
-import rpc_reader
-from plugin_remote import DeviceManager, MediaManager, SystemManager
-from scrypted_python.scrypted_sdk import ScryptedInterface, ScryptedStatic
+from scrypted_client import connect_scrypted_client
+from scrypted_python.scrypted_sdk import ScryptedInterface
 
 
-class EioRpcTransport(rpc_reader.RpcTransport):
-    def __init__(self, loop: asyncio.AbstractEventLoop):
-        super().__init__()
-        self.eio = engineio.AsyncClient(ssl_verify=False)
-        self.loop = loop
-        self.write_error: Exception = None
-        self.read_queue = asyncio.Queue()
-        self.write_queue = asyncio.Queue()
-
-        @self.eio.on("message")
-        def on_message(data):
-            self.read_queue.put_nowait(data)
-
-        asyncio.run_coroutine_threadsafe(self.send_loop(), self.loop)
-
-    async def read(self):
-        return await self.read_queue.get()
-
-    async def send_loop(self):
-        while True:
-            data = await self.write_queue.get()
-            try:
-                await self.eio.send(data)
-            except Exception as e:
-                self.write_error = e
-                self.write_queue = None
-                break
-
-    def writeBuffer(self, buffer, reject):
-        async def send():
-            try:
-                if self.write_error:
-                    raise self.write_error
-                self.write_queue.put_nowait(buffer)
-            except Exception as e:
-                reject(e)
-
-        asyncio.run_coroutine_threadsafe(send(), self.loop)
-
-    def writeJSON(self, json, reject):
-        return self.writeBuffer(json, reject)
-
-
-async def connect_scrypted_client(
-    transport: EioRpcTransport,
-    base_url: str,
-    username: str,
-    password: str,
-    plugin_id: str = "@scrypted/core",
-    session: aiohttp.ClientSession | None = None,
-) -> ScryptedStatic:
-    login_url = f"{base_url}/login"
-    login_body = {
-        "username": username,
-        "password": password,
-    }
-
-    if session:
-        cm = nullcontext(session)
-    else:
-        cm = aiohttp.ClientSession()
-
-    async with cm as _session:
-        async with _session.post(
-            login_url, verify_ssl=False, json=login_body
-        ) as response:
-            login_response = await response.json()
-
-        headers = {"Authorization": login_response["authorization"]}
-
-        await transport.eio.connect(
-            base_url,
-            headers=headers,
-            engineio_path=f"/endpoint/{plugin_id}/engine.io/api/",
-        )
-
-        ret = asyncio.Future[ScryptedStatic](loop=transport.loop)
-        peer, peerReadLoop = await rpc_reader.prepare_peer_readloop(
-            transport.loop, transport
-        )
-        peer.params["print"] = print
-
-        def callback(api, pluginId, hostInfo):
-            remote = plugin_remote.PluginRemote(
-                peer, api, pluginId, hostInfo, transport.loop
-            )
-            wrapped = remote.setSystemState
-
-            async def remoteSetSystemState(systemState):
-                await wrapped(systemState)
-
-                async def resolve():
-                    sdk = ScryptedStatic()
-                    sdk.api = api
-                    sdk.remote = remote
-                    sdk.systemManager = SystemManager(api, remote.systemState)
-                    sdk.deviceManager = DeviceManager(
-                        remote.nativeIds, sdk.systemManager
-                    )
-                    sdk.mediaManager = MediaManager(await api.getMediaManager())
-                    ret.set_result(sdk)
-
-                asyncio.run_coroutine_threadsafe(resolve(), transport.loop)
-
-            remote.setSystemState = remoteSetSystemState
-            return remote
-
-        peer.params["getRemote"] = callback
-        asyncio.run_coroutine_threadsafe(peerReadLoop(), transport.loop)
-
-        sdk = await ret
-        return sdk
-
-
-async def main():
-    transport = EioRpcTransport(asyncio.get_event_loop())
-    sdk = await connect_scrypted_client(
-        transport,
-        "https://localhost:10443",
+async def main(loop: asyncio.AbstractEventLoop):
+    transport, sdk = await connect_scrypted_client(
+        loop,
+        os.environ.get("SCRYPTED_BASE_URL", "https://localhost:10443"),
         os.environ["SCRYPTED_USERNAME"],
         os.environ["SCRYPTED_PASSWORD"],
     )
@@ -142,10 +21,12 @@ async def main():
         if ScryptedInterface.OnOff.value in device.interfaces:
             print(f"OnOff: device is {device.on}")
 
-    await transport.eio.disconnect()
-    os._exit(0)
+    await transport.close()
 
 
-loop = asyncio.new_event_loop()
-asyncio.run_coroutine_threadsafe(main(), loop)
-loop.run_forever()
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main(loop))
+    finally:
+        loop.close()

server/python/plugin_remote.py

@@ -20,14 +20,10 @@ from io import StringIO
 from pathlib import Path
 from typing import Any, Callable, Coroutine, Optional, Set, Tuple, TypedDict
 
-import cluster_labels
-import plugin_console
-import plugin_volume as pv
 import rpc
 import rpc_reader
 import scrypted_python.scrypted_sdk.types
 from cluster_setup import ClusterSetup
-from plugin_pip import install_with_pip, need_requirements, remove_pip_dirs
 from scrypted_python.scrypted_sdk import PluginFork, ScryptedStatic
 from scrypted_python.scrypted_sdk.types import (Device, DeviceManifest,
                                                 EventDetails,
@@ -723,6 +719,13 @@ class PluginRemote:
             raise
 
     async def loadZipWrapped(self, packageJson, zipAPI: Any, zipOptions: dict):
+        # plugin host modules, unused (and not installed) when this module is
+        # consumed by a client via the scrypted-sdk package
+        import cluster_labels
+        import plugin_console
+        import plugin_volume as pv
+        from plugin_pip import install_with_pip, need_requirements, remove_pip_dirs
+
         await self.clusterSetup.initializeCluster(zipOptions)
 
         sdk = ScryptedStatic()