| """ |
| Configuration management for ARF Demo |
| Updated with REAL ARF installation detection |
| """ |
| from typing import Optional, Dict, Any, List |
| from enum import Enum |
| import os |
| import logging |
| import sys |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| try: |
| from pydantic_settings import BaseSettings |
| from pydantic import Field, validator |
| PYDANTIC_V2 = True |
| logger.info("Using pydantic-settings for BaseSettings") |
| except ImportError: |
| try: |
| from pydantic import BaseSettings, Field, validator |
| PYDANTIC_V2 = False |
| logger.info("Using pydantic.BaseSettings (older version)") |
| except ImportError as e: |
| logger.error(f"Failed to import pydantic: {e}") |
| |
| class BaseSettings: |
| def __init__(self, **kwargs): |
| for k, v in kwargs.items(): |
| setattr(self, k, v) |
| |
| class Field: |
| @staticmethod |
| def default(value): |
| return value |
| |
| def validator(*args, **kwargs): |
| def decorator(func): |
| return func |
| return decorator |
| |
| PYDANTIC_V2 = False |
|
|
|
|
| class ARFMode(str, Enum): |
| """ARF operation modes""" |
| DEMO = "demo" |
| OSS = "oss" |
| ENTERPRISE = "enterprise" |
|
|
|
|
| class SafetyMode(str, Enum): |
| """Safety modes for execution""" |
| ADVISORY = "advisory" |
| APPROVAL = "approval" |
| AUTONOMOUS = "autonomous" |
|
|
|
|
| class InstallationStatus(str, Enum): |
| """ARF package installation status""" |
| NOT_INSTALLED = "not_installed" |
| OSS_ONLY = "oss_only" |
| ENTERPRISE = "enterprise" |
| BOTH = "both" |
|
|
|
|
| class Settings(BaseSettings): |
| """ |
| Application settings with environment variable support |
| """ |
| |
| |
| arf_mode: ARFMode = Field( |
| default=ARFMode.DEMO, |
| description="ARF operation mode" |
| ) |
| |
| use_true_arf: bool = Field( |
| default=True, |
| description="Use true ARF integration when available" |
| ) |
| |
| |
| arf_oss_installed: bool = Field( |
| default=False, |
| description="ARF OSS package installed" |
| ) |
| |
| arf_enterprise_installed: bool = Field( |
| default=False, |
| description="ARF Enterprise package installed" |
| ) |
| |
| arf_oss_version: Optional[str] = Field( |
| default=None, |
| description="ARF OSS version if installed" |
| ) |
| |
| arf_enterprise_version: Optional[str] = Field( |
| default=None, |
| description="ARF Enterprise version if installed" |
| ) |
| |
| |
| arf_api_key: Optional[str] = Field( |
| default=None, |
| description="ARF API key for real integration" |
| ) |
| |
| arf_base_url: str = Field( |
| default="https://api.arf.dev", |
| description="ARF API base URL" |
| ) |
| |
| |
| engineer_hourly_rate: float = Field( |
| default=150.0, |
| description="Engineer hourly rate in USD" |
| ) |
| |
| engineer_annual_cost: float = Field( |
| default=125000.0, |
| description="Engineer annual cost in USD" |
| ) |
| |
| default_savings_rate: float = Field( |
| default=0.82, |
| description="Default savings rate with ARF" |
| ) |
| |
| |
| auto_refresh_seconds: int = Field( |
| default=30, |
| description="Auto-refresh interval in seconds" |
| ) |
| |
| max_history_items: int = Field( |
| default=100, |
| description="Maximum history items to display" |
| ) |
| |
| |
| default_scenario: str = Field( |
| default="Cache Miss Storm", |
| description="Default incident scenario" |
| ) |
| |
| scenario_config_path: str = Field( |
| default="config/scenarios", |
| description="Path to scenario configuration files" |
| ) |
| |
| |
| default_safety_mode: SafetyMode = Field( |
| default=SafetyMode.ADVISORY, |
| description="Default safety mode" |
| ) |
| |
| require_approval: bool = Field( |
| default=True, |
| description="Require human approval for execution" |
| ) |
| |
| |
| @validator("arf_api_key") |
| def validate_api_key(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]: |
| if values.get("arf_mode") == ARFMode.ENTERPRISE and not v: |
| raise ValueError("ARF API key required for Enterprise mode") |
| return v |
| |
| @validator("use_true_arf") |
| def validate_true_arf(cls, v: bool, values: Dict[str, Any]) -> bool: |
| if v and not values.get("arf_oss_installed"): |
| logger.warning("True ARF requested but OSS package not installed. Using mock mode.") |
| return False |
| return v |
| |
| |
| @classmethod |
| def detect_installation(cls): |
| """Detect ARF package installation""" |
| results = { |
| "oss_installed": False, |
| "enterprise_installed": False, |
| "oss_version": None, |
| "enterprise_version": None |
| } |
| |
| |
| try: |
| import agentic_reliability_framework as arf_oss |
| results["oss_installed"] = True |
| results["oss_version"] = getattr(arf_oss, '__version__', '3.3.7') |
| logger.info(f"β
ARF OSS v{results['oss_version']} detected") |
| except ImportError: |
| logger.info("β οΈ ARF OSS not installed - will use mock mode") |
| |
| |
| try: |
| import arf_enterprise |
| results["enterprise_installed"] = True |
| results["enterprise_version"] = getattr(arf_enterprise, '__version__', '1.0.2') |
| logger.info(f"β
ARF Enterprise v{results['enterprise_version']} detected") |
| except ImportError: |
| logger.info("β οΈ ARF Enterprise not installed - will use simulation") |
| |
| return results |
| |
| def get_installation_status(self) -> InstallationStatus: |
| """Get current installation status""" |
| if self.arf_oss_installed and self.arf_enterprise_installed: |
| return InstallationStatus.BOTH |
| elif self.arf_enterprise_installed: |
| return InstallationStatus.ENTERPRISE |
| elif self.arf_oss_installed: |
| return InstallationStatus.OSS_ONLY |
| else: |
| return InstallationStatus.NOT_INSTALLED |
| |
| def get_installation_badges(self) -> Dict[str, Any]: |
| """Get badge information for UI display""" |
| if self.arf_oss_installed: |
| oss_badge = { |
| "text": f"β
ARF OSS v{self.arf_oss_version}", |
| "color": "#10b981", |
| "icon": "β
" |
| } |
| else: |
| oss_badge = { |
| "text": "β οΈ Mock ARF", |
| "color": "#f59e0b", |
| "icon": "β οΈ" |
| } |
| |
| if self.arf_enterprise_installed: |
| enterprise_badge = { |
| "text": f"π Enterprise v{self.arf_enterprise_version}", |
| "color": "#8b5cf6", |
| "icon": "π" |
| } |
| else: |
| enterprise_badge = { |
| "text": "π Enterprise Required", |
| "color": "#64748b", |
| "icon": "π" |
| } |
| |
| return { |
| "oss": oss_badge, |
| "enterprise": enterprise_badge |
| } |
| |
| def get_installation_recommendations(self) -> List[str]: |
| """Get installation recommendations""" |
| recommendations = [] |
| |
| if not self.arf_oss_installed: |
| recommendations.append( |
| "Install real ARF OSS: `pip install agentic-reliability-framework==3.3.7`" |
| ) |
| |
| if not self.arf_enterprise_installed: |
| recommendations.append( |
| "Install ARF Enterprise: `pip install agentic-reliability-enterprise` (requires license)" |
| ) |
| |
| return recommendations |
| |
| class Config: |
| env_file = ".env" |
| env_file_encoding = "utf-8" |
| case_sensitive = False |
| use_enum_values = True |
|
|
|
|
| |
| try: |
| |
| installation_info = Settings.detect_installation() |
| |
| |
| settings = Settings( |
| arf_oss_installed=installation_info["oss_installed"], |
| arf_enterprise_installed=installation_info["enterprise_installed"], |
| arf_oss_version=installation_info["oss_version"], |
| arf_enterprise_version=installation_info["enterprise_version"], |
| ) |
| |
| |
| status = settings.get_installation_status() |
| logger.info(f"ARF Installation Status: {status.value}") |
| |
| if status == InstallationStatus.NOT_INSTALLED: |
| logger.warning("No ARF packages installed. Demo will use mock mode.") |
| logger.warning("For real ARF experience, install: pip install agentic-reliability-framework==3.3.7") |
| |
| except Exception as e: |
| logger.warning(f"Failed to load settings: {e}, using defaults") |
| settings = Settings( |
| arf_mode=ARFMode.DEMO, |
| use_true_arf=False, |
| arf_oss_installed=False, |
| arf_enterprise_installed=False, |
| engineer_hourly_rate=150.0, |
| engineer_annual_cost=125000.0, |
| default_savings_rate=0.82, |
| auto_refresh_seconds=30, |
| max_history_items=100, |
| default_scenario="Cache Miss Storm", |
| scenario_config_path="config/scenarios", |
| default_safety_mode=SafetyMode.ADVISORY, |
| require_approval=True |
| ) |
|
|
|
|
| def get_settings() -> Settings: |
| """Get settings instance (singleton pattern)""" |
| return settings |
|
|
|
|
| def print_installation_status(): |
| """Print installation status to console""" |
| s = get_settings() |
| status = s.get_installation_status() |
| |
| print("=" * 70) |
| print("π ARF Ultimate Investor Demo - Installation Status") |
| print("=" * 70) |
| |
| print(f"π¦ ARF OSS: {'β
v' + s.arf_oss_version if s.arf_oss_installed else 'β οΈ Not installed'}") |
| print(f"π’ Enterprise: {'β
v' + s.arf_enterprise_version if s.arf_enterprise_installed else 'β οΈ Not installed'}") |
| print(f"π― Mode: {s.arf_mode.value}") |
| print(f"π€ Using True ARF: {'β
Yes' if s.use_true_arf else 'β οΈ Mock mode'}") |
| |
| recommendations = s.get_installation_recommendations() |
| if recommendations: |
| print("\nπ‘ Recommendations:") |
| for rec in recommendations: |
| print(f" β’ {rec}") |
| |
| print("=" * 70) |
|
|
|
|
| |
| if __name__ != "__main__" and "pytest" not in sys.modules: |
| print_installation_status() |