πŸ“„ config.py 12,973 bytes Apr 19, 2026 πŸ“‹ Raw

"""Centralized configuration for the Family Assistant package.

All env var reads and defaults live here. No other module reads os.environ
directly for these values.
"""

import os
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

import yaml

---------------------------------------------------------------------------

.env auto-load (before reading any env vars)

---------------------------------------------------------------------------

If a .env file exists in CWD or the scripts directory, load it into os.environ.

This ensures cron/heartbeat sessions have access to config even without .bashrc.

ENV_SEARCH_PATHS = [Path("."), Path(__file__).resolve().parent.parent]
for _p in _ENV_SEARCH_PATHS:
_env_file = _p / ".env"
if _env_file.is_file():
for _line in _env_file.read_text().splitlines():
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_key,
, _val = _line.partition("=")
_key = _key.strip()
_val = _val.strip()
if _key not in os.environ: # don't override existing env vars
os.environ[_key] = _val
break

---------------------------------------------------------------------------

Gmail / IMAP (DORMANT β€” kept as fallback, not primary ingress)

---------------------------------------------------------------------------

GMAIL_USER = os.environ.get("GMAIL_USER", "")
GMAIL_APP_PASSWORD = os.environ.get("GMAIL_APP_PASSWORD")
IMAP_SERVER = "imap.gmail.com"
IMAP_PORT = 993

---------------------------------------------------------------------------

CalDAV (Radicale on Beelink β€” PRIMARY calendar backend)

---------------------------------------------------------------------------

CALDAV_URL = os.environ.get("CALDAV_URL", "http://127.0.0.1:5232")
CALDAV_USER = os.environ.get("CALDAV_USER", "assistant")
CALDAV_PASSWORD = os.environ.get("CALDAV_PASSWORD", "") # Required, no default
CALDAV_CALENDAR_NAME = os.environ.get("CALDAV_CALENDAR_NAME", "family")

Legacy Google Calendar vars (kept for backward compat, not used by calendar_sync)

GCAL_CALENDAR_ID = os.environ.get("GCAL_CALENDAR_ID", "")
GCAL_KEY_FILE = os.environ.get("GCAL_SERVICE_ACCOUNT", "")
SCOPES = [] # No Google scopes needed

---------------------------------------------------------------------------

LLM

---------------------------------------------------------------------------

LLM_URL = os.environ.get(
"LLM_URL", "http://localhost:11434/v1/chat/completions"
)
LLM_MODEL = os.environ.get("LLM_MODEL", "qwen2.5-coder:7b")

Newsletter extraction needs a stronger model β€” falls back to LLM_MODEL if not set

LLM_NEWSLETTER_MODEL = os.environ.get(
"LLM_NEWSLETTER_MODEL", "qwen2.5-coder:7b"
)

URL for newsletter model (override via .env; defaults to localhost)

LLM_NEWSLETTER_URL = os.environ.get(
"LLM_NEWSLETTER_URL", os.environ.get("LLM_URL", "http://localhost:11434/v1/chat/completions")
)
LLM_TIMEOUT = 30 # seconds
LLM_RESOLVE_TIMEOUT = 60 # seconds β€” resolution needs more thinking
LLM_NEWSLETTER_TIMEOUT = 180 # seconds β€” newsletters are long, dedup needs time

---------------------------------------------------------------------------

Timezone & limits

---------------------------------------------------------------------------

CHICAGO_TZ = ZoneInfo("America/Chicago")
MAX_BODY_CHARS = 1000
MIN_OVERLAP_MINUTES = 5

---------------------------------------------------------------------------

Family config

---------------------------------------------------------------------------

FAMILY_CONFIG_PATH = os.environ.get(
"FAMILY_CONFIG_PATH", ""
)

Cached family config after first load

_family_config = None

def _resolve_family_config_path():
"""Resolve the family.yaml path using FAMILY_CONFIG_PATH, CWD, or XDG config."""
if FAMILY_CONFIG_PATH:
p = Path(FAMILY_CONFIG_PATH)
if p.exists():
return p
# Try expanding ~
p = Path(os.path.expanduser(FAMILY_CONFIG_PATH))
if p.exists():
return p

# Try CWD
cwd_path = Path.cwd() / "family.yaml"
if cwd_path.exists():
    return cwd_path

# Try XDG config dir
xdg_path = Path.home() / ".config" / "family-assistant" / "family.yaml"
if xdg_path.exists():
    return xdg_path

return None

def load_family_config():
"""Read the family.yaml config file and return the parsed dict.

Caches after first successful load. Raises FileNotFoundError if no
family.yaml can be found anywhere.
"""
global _family_config
if _family_config is not None:
    return _family_config

config_path = _resolve_family_config_path()
if config_path is None:
    raise FileNotFoundError(
        "No family.yaml found. Set FAMILY_CONFIG_PATH or run "
        "'family-assistant setup' to create one."
    )

with open(config_path, "r") as f:
    _family_config = yaml.safe_load(f)

if not _family_config or "family" not in _family_config:
    raise ValueError(f"family.yaml at {config_path} is missing the 'family' key")

return _family_config

def get_family_members_str():
"""Format family members for prompt injection.

Returns a string like:
  "Alex (dad), Jordan (mom), Charlie (son, child β€” needs adult),
   Sam (daughter, child β€” needs adult), Rover (dog β€” needs adult)"
"""
config = load_family_config()
members = config.get("family", {}).get("members", [])
parts = []
for m in members:
    name = m["name"]
    role = m.get("role", "")
    needs_adult = m.get("needs_adult", False)
    is_child = role.lower() in ("son", "daughter", "child")

    detail_parts = []
    if role:
        detail_parts.append(role)
    if is_child and needs_adult:
        detail_parts.append("child β€” needs adult")
    elif needs_adult:
        detail_parts.append("needs adult")

    if detail_parts:
        parts.append(f"{name} ({', '.join(detail_parts)})")
    else:
        parts.append(name)

return ", ".join(parts)

def get_nickname_map():
"""Return a dict mapping lowercase nicknames to canonical names.

E.g. {"chuck": "Charlie"} from a member with nicknames: [Chuck].
"""
config = load_family_config()
members = config.get("family", {}).get("members", [])
nick_map = {}
for m in members:
    name = m["name"]
    for nick in m.get("nicknames", []):
        nick_map[nick.lower()] = name
return nick_map

def _compute_current_grade(member):
"""Compute the current grade string for a child member.

Uses deterministic math from baseline_grade, baseline_year, and
advancement_month. Falls back to school_grade if baseline fields
are missing (backward compat).

Grade mapping: -2=Pre-Pre-K, -1=Pre-K, 0=Kindergarten, 1=1st, 2=2nd, ...
"""
# Backward compat: if only school_grade is set, use it directly
if "baseline_grade" not in member:
    return member.get("school_grade", "")

baseline_grade = member["baseline_grade"]
baseline_year = member.get("baseline_year", 2025)
advancement_month = member.get("advancement_month", 9)

now = datetime.now(CHICAGO_TZ)
current_year = now.year
current_month = now.month

# How many school years have elapsed since baseline?
# A school year starts in advancement_month of baseline_year.
# If current date is past advancement_month of year X, we're in the X-X+1 school year.
if current_month >= advancement_month:
    school_years_elapsed = current_year - baseline_year
else:
    school_years_elapsed = current_year - baseline_year - 1

if school_years_elapsed < 0:
    school_years_elapsed = 0

current_grade_int = baseline_grade + school_years_elapsed
return _grade_int_to_string(current_grade_int)

def _grade_int_to_string(grade_int):
"""Convert an integer grade to a human-readable string.

-2 β†’ Pre-Pre-K, -1 β†’ Pre-K, 0 β†’ Kindergarten,
1 β†’ 1st, 2 β†’ 2nd, 3 β†’ 3rd, 4+ β†’ Nth (with proper ordinal suffix)
"""
if grade_int <= -2:
    return "Pre-Pre-K"
if grade_int == -1:
    return "Pre-K"
if grade_int == 0:
    return "Kindergarten"
# Ordinal suffix
if grade_int == 1:
    return "1st grade"
if grade_int == 2:
    return "2nd grade"
if grade_int == 3:
    return "3rd grade"
if 11 <= grade_int % 100 <= 13:
    return f"{grade_int}th grade"
if grade_int % 10 == 1:
    return f"{grade_int}st grade"
if grade_int % 10 == 2:
    return f"{grade_int}nd grade"
if grade_int % 10 == 3:
    return f"{grade_int}rd grade"
return f"{grade_int}th grade"

def get_children_profiles_str():
"""Format children's profiles for prompt injection.

Uses dynamic grade computation from baseline_grade/baseline_year/
advancement_month. Falls back to school_grade for backward compat.

Returns a string like:
  "Charlie (1st grade, sports: soccer/t-ball, clubs: none),
   Sam (Pre-K, sports: none, clubs: none)"
Returns empty string if no children have profile fields.
"""
config = load_family_config()
members = config.get("family", {}).get("members", [])
profiles = []
for m in members:
    role = m.get("role", "").lower()
    if role not in ("son", "daughter", "child"):
        continue
    name = m["name"]
    grade = _compute_current_grade(m)
    sports = m.get("sports", [])
    clubs = m.get("clubs", [])

    parts = []
    if grade:
        parts.append(grade)
    if sports:
        parts.append(f"sports: {', '.join(sports)}")
    else:
        parts.append("sports: none")
    if clubs:
        parts.append(f"clubs: {', '.join(clubs)}")
    else:
        parts.append("clubs: none")

    detail = ", ".join(parts)
    profiles.append(f"{name} ({detail})")

if not profiles:
    return ""
return "Children profiles: " + "; ".join(profiles)

def get_nickname_rules_str():
"""Format nickname rules for prompt injection.

E.g. "Normalize nicknames: Chuck→Charlie"
Returns empty string if no nicknames defined.
"""
config = load_family_config()
members = config.get("family", {}).get("members", [])
rules = []
for m in members:
    name = m["name"]
    for nick in m.get("nicknames", []):
        rules.append(f"{nick}β†’{name}")

if not rules:
    return ""
return "Normalize nicknames: " + ", ".join(rules)

---------------------------------------------------------------------------

Google Places (goplaces)

---------------------------------------------------------------------------

GOOGLE_PLACES_API_KEY = os.environ.get("GOOGLE_PLACES_API_KEY", "")

---------------------------------------------------------------------------

Google Drive (Drop-Box Document Sorter)

---------------------------------------------------------------------------

DRIVE_FOLDER_ID = os.environ.get("DRIVE_FOLDER_ID", "")
DRIVE_SA_KEY_PATH = os.environ.get(
"DRIVE_SA_KEY_PATH",
os.path.expanduser("~/.openclaw/secrets/gcal-service-account.json"),
)
VISION_LLM_URL = os.environ.get("VISION_LLM_URL", "")
VISION_MODEL = os.environ.get("VISION_MODEL", "qwen3-vl:8b")

---------------------------------------------------------------------------

Prompt loading

---------------------------------------------------------------------------

_PROMPTS_DIR = Path(file).parent / "prompts"

def load_prompts():
"""Read the prompt text files and return them as a dict.

Injects family-specific placeholders into prompts:
  {family_members} β€” formatted list of family members
  {nickname_rules} β€” nickname normalization instructions

Returns:
    dict with keys 'appointment_extract', 'appointment_retry', 'conflict_resolve'
"""
# Load family context for prompt injection
family_members = get_family_members_str()
nickname_rules = get_nickname_rules_str()
children_profiles = get_children_profiles_str()

prompts = {}
for name, filename in [
    ("appointment_extract", "appointment_extract.txt"),
    ("appointment_retry", "appointment_retry.txt"),
    ("conflict_resolve", "conflict_resolve.txt"),
    ("email_classify", "email_classify.txt"),
    ("newsletter_extract", "newsletter_extract.txt"),
    ("newsletter_dedup", "newsletter_dedup.txt"),
]:
    path = _PROMPTS_DIR / filename
    text = path.read_text()
    # Inject family placeholders
    text = text.replace("{family_members}", family_members)
    text = text.replace("{nickname_rules}", nickname_rules)
    text = text.replace("{children_profiles}", children_profiles)
    prompts[name] = text
return prompts