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)

← Back