import os def _env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.lower() in {"1", "true", "yes", "on"} def _env_int(name: str, default: int) -> int: value = os.getenv(name) if value is None: return default try: return int(value) except (TypeError, ValueError): return default def _env_list(name: str, default: list[str]) -> list[str]: value = os.getenv(name) if value is None: return default parsed = [item.strip() for item in value.split(",") if item.strip()] return parsed or default # Defaults assume the appliance-style deployment model where Marvis runs with # host networking and talks to sibling services over host loopback. PROMETHEUS_URL = os.getenv("MARVIS_PROMETHEUS_URL", "http://127.0.0.1:9090") PROMETHEUS_PREFIX = os.getenv("MARVIS_PROMETHEUS_PREFIX", "/prometheus") ALERTMANAGER_URL = os.getenv("MARVIS_ALERTMANAGER_URL", "http://127.0.0.1:9093") # PLS discovery defaults assume the local appliance exposes PLS via Traefik. PLS_BASE_URL = os.getenv("MARVIS_PLS_BASE_URL", "https://127.0.0.1/core/pls/api/1") PLS_USERNAME = os.getenv("MARVIS_PLS_USERNAME", "") PLS_PASSWORD = os.getenv("MARVIS_PLS_PASSWORD", "") PLS_AUTH_BACKEND = os.getenv("MARVIS_PLS_AUTH_BACKEND", "local") PLS_VERIFY_TLS = _env_bool("MARVIS_PLS_VERIFY_TLS", False) # Fluent Bit ingestion and retention. LOG_INGEST_ENABLED = _env_bool("MARVIS_LOG_INGEST_ENABLED", True) LOG_AUTO_CONFIGURE = _env_bool("MARVIS_LOG_AUTO_CONFIGURE", True) LOG_RECEIVER_BIND_HOST = os.getenv("MARVIS_LOG_RECEIVER_BIND_HOST", "0.0.0.0") LOG_RECEIVER_HOST = os.getenv("MARVIS_LOG_RECEIVER_HOST", "") LOG_RECEIVER_PORT = _env_int("MARVIS_LOG_RECEIVER_PORT", 5514) LOG_RECEIVER_FORMAT = os.getenv("MARVIS_LOG_RECEIVER_FORMAT", "json_lines") LOG_BUFFER_LINES = _env_int("MARVIS_LOG_BUFFER_LINES", 1000) LOG_PROCESS_BUFFER_LINES = _env_int("MARVIS_LOG_PROCESS_BUFFER_LINES", 500) LOG_SUBSCRIBER_BUFFER_LINES = _env_int("MARVIS_LOG_SUBSCRIBER_BUFFER_LINES", 500) LOG_ALERT_CONTEXT_BEFORE = _env_int("MARVIS_LOG_ALERT_CONTEXT_BEFORE", 5) LOG_ALERT_CONTEXT_AFTER = _env_int("MARVIS_LOG_ALERT_CONTEXT_AFTER", 5) LOG_ALERT_CONTEXT_DB_PATH = os.getenv("MARVIS_LOG_ALERT_CONTEXT_DB_PATH", "/app/data/marvis-alert-context.db") LOG_ALERT_CONTEXT_DB_MAX_ROWS = _env_int("MARVIS_LOG_ALERT_CONTEXT_DB_MAX_ROWS", 500) LOG_TRACE_BUFFER_LINES = _env_int("MARVIS_LOG_TRACE_BUFFER_LINES", 5000) LOG_FLUENTBIT_MATCH = os.getenv("MARVIS_LOG_FLUENTBIT_MATCH", "*") LOG_TRACE_DEBUG_LEVEL = os.getenv("MARVIS_LOG_TRACE_DEBUG_LEVEL", "debug") LOG_TRACE_TARGET_SERVICES = [item.lower() for item in _env_list( "MARVIS_LOG_TRACE_TARGET_SERVICES", ["amf", "smf", "mme", "upf", "sgwc"], )] LOG_ALLOWED_NFS = [item.upper() for item in _env_list( "MARVIS_LOG_ALLOWED_NFS", ["AMF", "SMF", "UPF", "UDM", "UDR", "NRF", "AUSF", "PCF", "MME", "SGWC", "DRA", "DSM", "AAA", "BMSC", "CHF", "SMSF", "EIR"], )] # AI backend: "rule" | "openai" | "ollama" AI_MODE = os.getenv("MARVIS_AI_MODE", "rule") OPENAI_API_KEY = os.getenv("MARVIS_OPENAI_API_KEY", "") OPENAI_MODEL = os.getenv("MARVIS_OPENAI_MODEL", "gpt-4o-mini") # Override to use any OpenAI-compatible local LLM (llama.cpp, vLLM, LM Studio, etc.) OPENAI_BASE_URL = os.getenv("MARVIS_OPENAI_BASE_URL", "https://api.openai.com") OLLAMA_URL = os.getenv("MARVIS_OLLAMA_URL", "http://127.0.0.1:11434") OLLAMA_MODEL = os.getenv("MARVIS_OLLAMA_MODEL", "llama3") # Container/runtime integration CONTAINER_RUNTIME = os.getenv("MARVIS_CONTAINER_RUNTIME", "docker") CONTAINER_HOST = os.getenv("CONTAINER_HOST", "") UERANSIM_IMAGE = os.getenv("MARVIS_UERANSIM_IMAGE", "p5g-marvis-ueransim") UERANSIM_ENV_FILE = os.getenv("MARVIS_UERANSIM_ENV_FILE", "/app/config/ueransim.env") UERANSIM_NETWORK_MODE = os.getenv("MARVIS_UERANSIM_NETWORK_MODE", "host") UERANSIM_PRIVILEGED = os.getenv("MARVIS_UERANSIM_PRIVILEGED", "true").lower() in {"1", "true", "yes", "on"} UERANSIM_CONTAINER_NAME = os.getenv("MARVIS_UERANSIM_CONTAINER_NAME", "marvis-ueransim-test") # Maps Prometheus target_type label → display name TARGET_TYPE_MAP = { "amf": "AMF", "smf": "SMF", "upf": "UPF", "udm": "UDM", "udr": "UDR", "nrf": "NRF", "ausf": "AUSF", "pcf": "PCF", "mme": "MME", "sgwc": "SGWC", "dra": "DRA", "dsm": "DSM", "ncm": "NCM", "pls": "PLS", } ALL_NFS = ["AMF", "SMF", "UPF", "UDM", "UDR", "NRF", "AUSF", "PCF", "MME", "SGWC", "DRA", "DSM"]