📄 family_loader.py 5,034 bytes Apr 26, 2026 📋 Raw

"""Family configuration loader — Single-tenant only.

Loads family data from YAML and provides pattern matching for inference.
No multi-tenant support — hardcoded to Hoffmann household.
"""

import re
import yaml
from pathlib import Path
from typing import Optional

SINGLE-TENANT: Hardcoded family

DEFAULT_FAMILY_ID = "hoffmann"

Find config relative to this file (icarus/core/family_loader.py → icarus/config/families/)

FAMILY_CONFIG_PATH = Path(file).parent.parent / "config" / "families" / f"{DEFAULT_FAMILY_ID}.yaml"

class FamilyConfig:
"""Single-tenant family configuration."""

def __init__(self, family_id: str = DEFAULT_FAMILY_ID):
    """Load family configuration from YAML.

    Args:
        family_id: Must be "hoffmann". Single-tenant constraint.
    """
    if family_id != DEFAULT_FAMILY_ID:
        raise ValueError(f"Single-tenant only. Family must be '{DEFAULT_FAMILY_ID}'.")

    self.family_id = family_id
    self._config = self._load_config()
    self._compiled_rules = self._compile_rules()

def _load_config(self) -> dict:
    """Load and parse YAML configuration."""
    if not FAMILY_CONFIG_PATH.exists():
        raise FileNotFoundError(f"Family config not found: {FAMILY_CONFIG_PATH}")

    with open(FAMILY_CONFIG_PATH) as f:
        return yaml.safe_load(f)

def _compile_rules(self) -> list:
    """Pre-compile regex patterns for fast matching."""
    compiled = []
    for rule in self.inference_rules:
        try:
            compiled.append({
                **rule,
                "_regex": re.compile(rule["pattern"], re.IGNORECASE)
            })
        except re.error as e:
            print(f"Warning: Invalid regex in rule {rule['id']}: {e}")
    return compiled

@property
def members(self) -> list[dict]:
    """Return all family members."""
    return self._config.get("members", [])

@property
def inference_rules(self) -> list[dict]:
    """Return pattern matching rules."""
    return self._config.get("inference_rules", [])

@property
def fallback_threshold(self) -> float:
    """Confidence threshold for asking user."""
    return self._config.get("telemetry", {}).get("fallback_threshold", 0.70)

def get_member(self, member_id: str) -> Optional[dict]:
    """Get member by ID."""
    for member in self.members:
        if member.get("id") == member_id:
            return member
    return None

def infer_recipients(self, text: str) -> list[dict]:
    """Infer which family members a document concerns.

    Uses deterministic pattern matching (no LLM).
    Returns: [{member_id, confidence, rule_id, description}]
    """
    text_lower = text.lower()
    matches = []
    matched_ids = set()

    for rule in self._compiled_rules:
        if rule["_regex"].search(text):
            for member_id in rule["assign_to"]:
                if member_id not in matched_ids:
                    matches.append({
                        "member_id": member_id,
                        "member": self.get_member(member_id),
                        "confidence": rule["confidence"],
                        "rule_id": rule["id"],
                        "description": rule["description"]
                    })
                    matched_ids.add(member_id)

    return matches

def build_context_prompt(self) -> str:
    """Build family context string for LLM prompts."""
    lines = ["FAMILY MEMBERS:"]

    for member in self.members:
        if member.get("current_grade"):  # Kids for school docs
            lines.append(f"- {member['name']} ({member.get('nickname', 'no nickname')}) Grade {member['current_grade']}, Teacher: {member.get('teacher', 'unknown')}")
        elif member.get("is_parent"):
            lines.append(f"- {member['name']} (parent)")

    lines.append("")
    lines.append("INFERENCE RULES:")
    for rule in self.inference_rules:
        lines.append(f"- {rule['description']} → {', '.join(rule['assign_to'])} (confidence: {rule['confidence']})")

    return "\n".join(lines)

def get_telemetry_config(self) -> dict:
    """Get telemetry/logging configuration."""
    return self._config.get("telemetry", {})

Singleton instance

_family_config: Optional[FamilyConfig] = None

def get_family_config() -> FamilyConfig:
"""Get singleton family configuration.

Single-tenant: Always returns Hoffmann family config.
"""
global _family_config
if _family_config is None:
    _family_config = FamilyConfig()
return _family_config

def reset_family_config():
"""Reset singleton (useful for testing)."""
global _family_config
_family_config = None