engine_info.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""Single discovery file: ~/.aquin/engine-info.json + AQUIN_ENGINE_PORT."""
from __future__ import annotations
 
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
 
# Product default — uncommon on purpose; never 8002.
DEFAULT_ENGINE_PORT = 17832
DEFAULT_ENGINE_HOST = "127.0.0.1"
ENGINE_INFO_VERSION = "1"
 
 
def aquin_home() -> Path:
    override = (os.environ.get("AQUIN_HOME") or "").strip()
    if override:
        return Path(override)
    return Path.home() / ".aquin"
 
 
def engine_info_path() -> Path:
    return aquin_home() / "engine-info.json"
 
 
def load_engine_info() -> dict[str, Any] | None:
    path = engine_info_path()
    if not path.is_file():
        return None
    raw = path.read_text(encoding="utf-8-sig")
    data = json.loads(raw)
    if not isinstance(data, dict):
        raise ValueError("engine-info.json is not an object")
    return data
 
 
def resolve_engine_port(info: dict[str, Any] | None = None) -> int:
    env = (os.environ.get("AQUIN_ENGINE_PORT") or "").strip()
    if env:
        port = int(env)
        if port <= 0 or port > 65535:
            raise ValueError(f"invalid AQUIN_ENGINE_PORT: {env}")
        return port
    data = load_engine_info() if info is None else info
    if data and data.get("port") is not None:
        port = int(data["port"])
        if 0 < port <= 65535:
            return port
    return DEFAULT_ENGINE_PORT
 
 
def resolve_engine_host(info: dict[str, Any] | None = None) -> str:
    data = load_engine_info() if info is None else info
    if data and isinstance(data.get("host"), str) and data["host"].strip():
        return data["host"].strip()
    return DEFAULT_ENGINE_HOST
 
 
def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
 
def write_engine_info(patch: dict[str, Any]) -> dict[str, Any]:
    """Merge patch into engine-info.json (create parents / defaults as needed)."""
    home = aquin_home()
    home.mkdir(parents=True, exist_ok=True)
    existing: dict[str, Any] = {}
    try:
        loaded = load_engine_info()
        if isinstance(loaded, dict):
            existing = loaded
    except Exception:
        existing = {}
 
    daemon = dict(existing.get("daemon") or {})
    if isinstance(patch.get("daemon"), dict):
        daemon.update(patch["daemon"])
 
    merged: dict[str, Any] = {
        "version": ENGINE_INFO_VERSION,
        "engine_version": patch.get("engine_version", existing.get("engine_version")),
        "python": patch.get("python") or existing.get("python") or sys.executable,
        "compute_ready": (
            int(patch["compute_ready"])
            if patch.get("compute_ready") is not None
            else int(existing.get("compute_ready", 0) or 0)
        ),
        "port": int(patch["port"]) if patch.get("port") is not None else resolve_engine_port(existing),
        "host": patch.get("host") or existing.get("host") or DEFAULT_ENGINE_HOST,
        "bin": patch.get("bin") or existing.get("bin") or str(home / "bin" / ("aquin-engine.cmd" if os.name == "nt" else "aquin-engine")),
        "lib": patch.get("lib") or existing.get("lib") or str(home / "lib"),
        "daemon": {
            "status": daemon.get("status", "stopped"),
            "pid": daemon.get("pid"),
            "started_at": daemon.get("started_at"),
            "model_id": daemon.get("model_id"),
        },
        "updated_at": _utc_now(),
    }
    path = engine_info_path()
    path.write_text(json.dumps(merged, indent=2) + "\n", encoding="utf-8")
    return merged
 
 
def set_daemon_state(
    *,
    status: str,
    pid: int | None = None,
    model_id: str | None = None,
    port: int | None = None,
    started_at: str | None = None,
) -> dict[str, Any]:
    return write_engine_info(
        {
            **({"port": port} if port is not None else {}),
            "daemon": {
                "status": status,
                "pid": pid,
                "started_at": started_at,
                "model_id": model_id,
            },
        }
    )