fix(image): import WanGP from pinned checkout

Author: georgi

combined/prepare_model.py

@@ -38,6 +38,17 @@ def send(self, payload: dict[str, object]) -> None:
             self._stream.flush()
 
 
+def _configure_wangp_root() -> Path:
+    """Make the pinned WanGP checkout importable from this copied script."""
+    root = Path(os.environ.get("WANGP_ROOT", "/opt/Wan2GP")).resolve()
+    if not (root / "shared").is_dir():
+        raise FileNotFoundError(f"WanGP checkout is missing its shared package: {root}")
+    root_text = str(root)
+    if root_text not in sys.path:
+        sys.path.insert(0, root_text)
+    return root
+
+
 def _model_files(runtime: Any, model_type: str) -> list[tuple[str, int, int]]:
     """Mirror WanGP's own pre-load selection, stopping before model loading."""
     module = runtime.module
@@ -135,6 +146,7 @@ def _prepare(runtime: Any, model_type: str) -> None:
 def main() -> int:
     emitter = _Emitter()
     request = _read_request()
+    wangp_root = _configure_wangp_root()
     model_type = str(request.get("model_type") or "").strip()
     if not model_type:
         raise ValueError("model_type is required")
@@ -188,7 +200,7 @@ def observe() -> None:
             from shared.api import init
 
             session = init(
-                root=os.environ.get("WANGP_ROOT", "/opt/Wan2GP"),
+                root=str(wangp_root),
                 config_path=os.environ.get("WANGP_CONFIG_PATH"),
                 output_dir=os.environ.get("WANGP_OUTPUT_DIR"),
                 console_output=True,

tests/test_prepare_model.py

@@ -4,7 +4,11 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
-from combined.prepare_model import _model_files, _prepare  # noqa: E402
+from combined.prepare_model import (  # noqa: E402
+    _configure_wangp_root,
+    _model_files,
+    _prepare,
+)
 
 
 class FakeWanGP:
@@ -45,6 +49,19 @@ def download_models(self, *args, **kwargs):
         self.downloads.append((*args, kwargs))
 
 
+def test_configure_wangp_root_adds_copied_checkout_to_import_path(
+    tmp_path: Path, monkeypatch
+) -> None:
+    (tmp_path / "shared").mkdir()
+    monkeypatch.setenv("WANGP_ROOT", str(tmp_path))
+    monkeypatch.setattr(sys, "path", list(sys.path))
+
+    root = _configure_wangp_root()
+
+    assert root == tmp_path.resolve()
+    assert sys.path[0] == str(tmp_path.resolve())
+
+
 def test_model_files_match_wangp_primary_secondary_and_modules() -> None:
     module = FakeWanGP()