Files
operating-room-monitor-server/backend/app/algo_host/bundle.py
2026-05-22 09:35:41 +08:00

60 lines
2.0 KiB
Python

"""Resolve reference algorithm bundle paths and default YAML (read-only; no vendor patches)."""
from __future__ import annotations
import copy
from pathlib import Path
from typing import Any
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_REFERENCE_BUNDLE_RELATIVE = "algorithm_subprocesses/5.15"
def configured_reference_bundle_relative() -> str:
from app.config import Settings
raw = (Settings().reference_bundle_relative or "").strip()
return raw or DEFAULT_REFERENCE_BUNDLE_RELATIVE
def default_reference_bundle_dir() -> Path:
raw = configured_reference_bundle_relative()
path = Path(raw).expanduser()
if path.is_absolute():
return path.resolve()
return (REPO_ROOT / path).resolve()
def resolve_reference_bundle_dir(bundle_dir: Path | None = None) -> Path:
if bundle_dir is None:
label = configured_reference_bundle_relative()
root = default_reference_bundle_dir()
else:
label = str(bundle_dir)
root = Path(bundle_dir).expanduser().resolve()
if not (root / "main.py").is_file():
raise FileNotFoundError(f"reference bundle main.py not found: {label} -> {root}")
if not (root / "code" / "repo_root.py").is_file():
raise FileNotFoundError(f"reference bundle vendor code not found: {label} -> {root / 'code'}")
return root
def load_reference_default_config(bundle_dir: Path | None = None) -> dict[str, Any]:
root = resolve_reference_bundle_dir(bundle_dir)
path = root / "configs" / "default_config.yaml"
data = yaml.safe_load(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise ValueError(f"invalid reference bundle default config: {path}")
return copy.deepcopy(data)
def resolve_bundle_relative_path(bundle_dir: Path, raw: str) -> Path:
p = Path((raw or "").strip())
if not str(p):
raise ValueError("empty bundle-relative path")
if p.is_absolute():
return p.resolve()
return (bundle_dir / p).resolve()