ria-toolkit-oss/src/ria_toolkit_oss/agent/config.py

"""Agent configuration stored at ``~/.ria/agent.json``.

Schema::

    {
        "hub_url": "https://riahub.example.com",
        "agent_id": "agent-abc123",
        "token": "rha_xxxx",
        "name": "lab-bench-1",
        "insecure": false,
        "tx_enabled": false,
        "tx_max_gain_db": null,
        "tx_max_duration_s": null,
        "tx_allowed_freq_ranges": null
    }
"""

from __future__ import annotations

import json
import os
from dataclasses import asdict, dataclass, field
from pathlib import Path


def _resolve_default_path() -> Path:
    return Path(os.environ.get("RIA_AGENT_CONFIG", str(Path.home() / ".ria" / "agent.json")))


@dataclass
class AgentConfig:
    hub_url: str = ""
    agent_id: str = ""
    token: str = ""
    name: str = ""
    insecure: bool = False
    api_key: str = ""
    tx_enabled: bool = False
    tx_max_gain_db: float | None = None
    tx_max_duration_s: float | None = None
    tx_allowed_freq_ranges: list[list[float]] | None = None
    extra: dict = field(default_factory=dict)


def default_path() -> Path:
    return _resolve_default_path()


def _coerce_ranges(raw) -> list[list[float]] | None:
    if raw is None:
        return None
    out: list[list[float]] = []
    for pair in raw:
        lo, hi = pair
        out.append([float(lo), float(hi)])
    return out


def load(path: Path | None = None) -> AgentConfig:
    p = path or _resolve_default_path()
    if not p.exists():
        return AgentConfig()
    data = json.loads(p.read_text())
    known = {f for f in AgentConfig.__dataclass_fields__ if f != "extra"}
    extra = {k: v for k, v in data.items() if k not in known}
    return AgentConfig(
        hub_url=data.get("hub_url", ""),
        agent_id=data.get("agent_id", ""),
        token=data.get("token", ""),
        name=data.get("name", ""),
        insecure=bool(data.get("insecure", False)),
        api_key=data.get("api_key", ""),
        tx_enabled=bool(data.get("tx_enabled", False)),
        tx_max_gain_db=(float(v) if (v := data.get("tx_max_gain_db")) is not None else None),
        tx_max_duration_s=(float(v) if (v := data.get("tx_max_duration_s")) is not None else None),
        tx_allowed_freq_ranges=_coerce_ranges(data.get("tx_allowed_freq_ranges")),
        extra=extra,
    )


def save(cfg: AgentConfig, path: Path | None = None) -> Path:
    p = path or _resolve_default_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    data = asdict(cfg)
    extra = data.pop("extra", {}) or {}
    data.update(extra)
    p.write_text(json.dumps(data, indent=2))
    os.chmod(p, 0o600)
    return p