main.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.
 
from __future__ import annotations
 
import json
import os
import re
from pathlib import Path
from typing import Any
 
_STATE_PATH = Path.home() / ".aquin" / "engine.json"
_BASE_URL_DEFAULT = "https://api.aquin.app"
_API_URL_RE = re.compile(r"https?://api\.aquin\.app(?:/\S*)?")
 
 
def scrub_user_message(text: str) -> str:
    """Strip internal API host from strings shown in the terminal."""
    return _API_URL_RE.sub("Aquin cloud", str(text))
 
 
def _base_url() -> str:
    return os.environ.get("AQUIN_BASE_URL", _BASE_URL_DEFAULT).rstrip("/")
 
 
def _load_state() -> dict:
    if _STATE_PATH.exists():
        try:
            return json.loads(_STATE_PATH.read_text())
        except Exception:
            pass
    return {}
 
 
def _save_state(data: dict) -> None:
    _STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
    _STATE_PATH.write_text(json.dumps(data, indent=2))
 
 
def resolve_api_key(*, allow_missing: bool = False) -> str:
    """Saved login (config.json / env) — never prompts interactively."""
    from aquin.auth_config import resolve_api_key as _resolve
    return _resolve(allow_missing=allow_missing)
 
 
def ensure_engine_auth_persisted() -> None:
    """Copy logged-in credentials into engine.json when missing."""
    state = _load_state()
    api_key = resolve_api_key(allow_missing=True)
    base_url = (state.get("base_url") or _base_url()).rstrip("/")
    patch: dict[str, Any] = {}
    if api_key and state.get("api_key") != api_key:
        patch["api_key"] = api_key
    if state.get("base_url") != base_url:
        patch["base_url"] = base_url
    if patch:
        _save_state({**state, **patch})
 
 
def format_engine_status_lines() -> list[str]:
    """Account-adjacent engine state for `aquin status`."""
    from aquin.compute.model_loader import get_active_model_id
    from aquin.session_mode import mode_for_active_model, mode_label
 
    state = _load_state()
    model_id = get_active_model_id()
    mode = mode_for_active_model()
 
    lines: list[str] = []
    if model_id:
        lines.append(f"Model:  {model_id}")
        if mode:
            lines.append(f"  mode: {mode_label(mode)}")
    else:
        lines.append("Model:  (none — run: aquin load --model <id>)")
 
    if state.get("device"):
        lines.append(f"Device: {state.get('device')}")
    loc = state.get("location") or {}
    if isinstance(loc, dict) and loc.get("location_label"):
        lines.append(f"  location: {loc['location_label']}")
    gpu_info = state.get("gpu_info")
    if gpu_info:
        try:
            parsed = json.loads(gpu_info) if isinstance(gpu_info, str) else gpu_info
            gpus = parsed.get("gpus") or []
            if gpus:
                g = gpus[0]
                lines.append(f"  gpu: {g.get('name', '?')} ({g.get('vram_gb', '?')} GB)")
        except Exception:
            pass
    return lines