""" Configuration loading for second-brain. Settings are read from config/settings.toml (or a path set via SECOND_BRAIN_CONFIG). All values have sensible defaults so the system works out of the box. The relational store is Postgres (petalbrain DB, `second_brain` schema). Connection details are resolved by `second_brain.database` — see that module for the URL resolution order. The vault is still a file-based Obsidian directory; this project does NOT dual-write to petalbrain.wiki. """ from __future__ import annotations import os from pathlib import Path from typing import Any try: import tomllib # Python 3.11+ except ModuleNotFoundError: import tomli as tomllib # type: ignore[no-reuse-impl] # --------------------------------------------------------------------------- # Defaults # --------------------------------------------------------------------------- _DEFAULTS: dict[str, Any] = { "vault_path": "~/Documents/second-brain-vault", "media_dir": "~/.local/share/second-brain/media", "subtitles_dir": "~/.local/share/second-brain/subtitles", "domains": ["development", "content", "business", "homelab"], "whisper": { # Tower default: faster-whisper large-v3 on cuda/float16. The # transcribe helper picks int8 automatically if the device is cpu. "whisper_model": "large-v3", "whisper_device": "cuda", "whisper_compute_type": "", }, "database": { # Empty by default — the runtime expects SECOND_BRAIN_DATABASE_URL # (or HERBYLAB_DATABASE_URL) in the environment. Fill this only if # you prefer keeping the URL inside settings.toml. "url": "", }, "extractor": { # "cli" runs `claude -p` under the Max OAuth subscription (default). # "api" uses the Anthropic SDK and requires ANTHROPIC_API_KEY. "backend": "cli", "model": "claude-sonnet-4-6", "cli_binary": "claude", "timeout_sec": 600, }, "embeddings": { # Best-effort embedding of extraction summaries into public.embeddings. # If disabled, or if Ollama / the DB is unreachable, the pipeline # logs a warning and keeps going — file/relational data stays the # source of truth. "enabled": True, # Default targets the `ollama` service on the homelab docker network # (same convention vault-mcp uses). For host-side runs, set # OLLAMA_URL=http://127.0.0.1:11434 in the env instead. "ollama_url": "http://ollama:11434", "model": "nomic-embed-text", "embed_field": "summary", }, "scheduler": { "window_start": "22:00", "window_end": "06:00", # CLI backend: throttle by request rate (no per-token billing). "max_calls_per_hour": 30, "min_gap_seconds": 120, # Legacy API-backend budget; kept for backwards compat. "max_tokens_per_hour": 100_000, }, } DOMAINS = ["development", "content", "business", "homelab"] def _merge(default: dict, override: dict) -> dict: """Shallow merge — override wins per top-level key.""" out = dict(default) out.update(override or {}) return out class Config: """Holds resolved configuration values.""" def __init__(self, raw: dict[str, Any]) -> None: self._raw = raw # --- paths --- @property def vault_path(self) -> Path: return Path(self._raw.get("vault_path", _DEFAULTS["vault_path"])).expanduser() @property def media_dir(self) -> Path: return Path(self._raw.get("media_dir", _DEFAULTS["media_dir"])).expanduser() @property def subtitles_dir(self) -> Path: return Path( self._raw.get("subtitles_dir", _DEFAULTS["subtitles_dir"]) ).expanduser() # --- transcription --- @property def whisper(self) -> dict[str, Any]: """Merged [whisper] block. Env vars in `transcribe.resolve_settings` still win over these for the tower worker.""" return _merge(_DEFAULTS["whisper"], self._raw.get("whisper", {})) @property def whisper_model(self) -> str: """Backwards-compat shortcut — older callers (and settings.toml in the wild) reference whisper_model at the top level.""" return ( self._raw.get("whisper_model") or self.whisper.get("whisper_model", "large-v3") ) # --- extraction --- @property def domains(self) -> list[str]: return self._raw.get("domains", _DEFAULTS["domains"]) @property def anthropic_api_key(self) -> str | None: return os.getenv("ANTHROPIC_API_KEY") # --- structured sections --- @property def database(self) -> dict[str, Any]: return _merge(_DEFAULTS["database"], self._raw.get("database", {})) @property def extractor(self) -> dict[str, Any]: return _merge(_DEFAULTS["extractor"], self._raw.get("extractor", {})) @property def embeddings(self) -> dict[str, Any]: return _merge(_DEFAULTS["embeddings"], self._raw.get("embeddings", {})) @property def scheduler(self) -> dict[str, Any]: return _merge(_DEFAULTS["scheduler"], self._raw.get("scheduler", {})) # --- prompts dir --- @property def prompts_dir(self) -> Path: """Directory containing per-domain prompt templates.""" # src/second_brain/config.py → ../../.. = project root here = Path(__file__).resolve().parent.parent.parent return here / "prompts" def ensure_dirs(self) -> None: """Create runtime directories if they don't exist.""" for d in (self.media_dir, self.subtitles_dir): d.mkdir(parents=True, exist_ok=True) # --------------------------------------------------------------------------- # Loader # --------------------------------------------------------------------------- _config_instance: Config | None = None def load_config(path: Path | None = None) -> Config: """Load (and cache) configuration from a TOML file.""" global _config_instance if _config_instance is not None: return _config_instance if path is None: env_path = os.getenv("SECOND_BRAIN_CONFIG") if env_path: path = Path(env_path) else: # Walk up from CWD looking for config/settings.toml cwd = Path.cwd() for candidate in [cwd / "config" / "settings.toml", cwd / "settings.toml"]: if candidate.exists(): path = candidate break raw: dict[str, Any] = {} if path and path.exists(): with open(path, "rb") as fh: raw = tomllib.load(fh) _config_instance = Config(raw) return _config_instance def reset_config_singleton() -> None: """Test hook — clear the cached Config so a re-read picks up new settings.""" global _config_instance _config_instance = None __all__ = ["Config", "DOMAINS", "load_config", "reset_config_singleton"]