"""Family configuration loader — reads YAML from config/families/.""" import os import re from pathlib import Path from typing import Optional import yaml # Default data directory — overridden by ICARUS_DATA_DIR env var DEFAULT_DATA_DIR = Path.home() / ".icarus" def get_data_dir() -> Path: """Return the data directory for Icarus state (DBs, logs, etc.).""" return Path(os.environ.get("ICARUS_DATA_DIR", str(DEFAULT_DATA_DIR))) # --------------------------------------------------------------------------- # Family config model # --------------------------------------------------------------------------- class FamilyConfig: """Loaded family configuration with member profiles and inference rules.""" def __init__(self, data: dict): self.family_id: str = data.get("family_id", "default") self.version: str = data.get("version", "0.1.0") self.last_updated: str = data.get("last_updated", "") self.members: list[dict] = data.get("members", []) self.inference_rules: list[dict] = data.get("inference_rules", []) self.telemetry: dict = data.get("telemetry", {}) # Build lookup maps self._member_by_id: dict[str, dict] = {m["id"]: m for m in self.members} self._member_by_name: dict[str, dict] = {} self._compiled_rules: list[tuple[re.Pattern, dict]] = [] for m in self.members: name = m.get("name", "").lower() nick = (m.get("nickname") or "").lower() if name: self._member_by_name[name] = m if nick: self._member_by_name[nick] = m for rule in self.inference_rules: try: pattern = re.compile(rule["pattern"], re.IGNORECASE) self._compiled_rules.append((pattern, rule)) except re.error as e: pass # skip invalid rules def get_member(self, member_id: str) -> Optional[dict]: return self._member_by_id.get(member_id) def find_member_by_name(self, name: str) -> Optional[dict]: return self._member_by_name.get(name.lower()) def all_parents(self) -> list[dict]: return [m for m in self.members if m.get("is_parent")] def all_children(self) -> list[dict]: return [m for m in self.members if not m.get("is_parent")] def match_inference_rules(self, text: str) -> list[tuple[str, float]]: """Run inference rules against text, return [(assigned_id, confidence), ...].""" results = [] for pattern, rule in self._compiled_rules: if pattern.search(text): confidence = rule.get("confidence", 0.7) for assignee in rule.get("assign_to", []): results.append((assignee, confidence)) return results @property def fallback_threshold(self) -> float: return self.telemetry.get("fallback_threshold", 0.70) # --------------------------------------------------------------------------- # Loader # --------------------------------------------------------------------------- def _find_family_yaml() -> Optional[Path]: """Search known locations for family config YAML.""" search_paths = [ Path(os.environ.get("ICARUS_FAMILY_CONFIG", "")), Path.cwd() / "config" / "families", Path(__file__).parent / "families", get_data_dir() / "families", ] for base in search_paths: if base.name and not base.exists(): continue if base.is_file() and base.suffix in (".yaml", ".yml"): return base if base.is_dir(): # Pick the first .yaml file for f in sorted(base.iterdir()): if f.suffix in (".yaml", ".yml"): return f return None def load_family(path: Optional[str] = None) -> FamilyConfig: """Load family configuration from YAML file. Resolution order: 1. Explicit path argument 2. ICARUS_FAMILY_CONFIG env var 3. ./config/families/*.yaml 4. ~/.icarus/families/*.yaml """ if path: yaml_path = Path(path) else: found = _find_family_yaml() if not found: raise FileNotFoundError( "No family config found. Create a YAML file in config/families/ " "or set ICARUS_FAMILY_CONFIG." ) yaml_path = found with open(yaml_path) as f: data = yaml.safe_load(f) return FamilyConfig(data)