"""Short-term structured memory for LifeOS. One shared JSON store at data/memory.json: meals: [{date, dish, ingredients[]}] workouts: [{date, type, duration_min}] finances: {income_monthly, subscriptions: [{name, cost, last_used}]} user_profile: {name, dietary_prefs, fitness_goal, budget_weekly} A real install starts blank (empty_data). Set LIFEOS_DEMO=1 to instead seed a sample persona whose dates are generated relative to "today" so the demo never goes stale. memory.json is gitignored and created on first import / app start. """ import json import os import threading import uuid from datetime import date, timedelta import config DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") MEMORY_PATH = os.path.join(DATA_DIR, "memory.json") _lock = threading.Lock() def _d(days_ago: int) -> str: return (date.today() - timedelta(days=days_ago)).isoformat() def _new_id() -> str: return uuid.uuid4().hex[:8] def _week_day(weekday: int) -> str: """Date of the given weekday (0=Mon) in the current week.""" today = date.today() return (today + timedelta(days=weekday - today.weekday())).isoformat() def _months_out(months: int) -> str: """YYYY-MM roughly `months` months from today.""" y, m = date.today().year, date.today().month + months y += (m - 1) // 12 m = (m - 1) % 12 + 1 return f"{y:04d}-{m:02d}" def empty_data() -> dict: """A blank store for a brand-new user. Same schema as seed_data() but with no personal history — the user fills it in via the Profile/Memory UI.""" return { "meals": [], "workouts": [], "calendar": [], "workout_schedule": {"days": [], "time": "07:00"}, "goals": [], "finances": { "income_monthly": 0, "monthly_payments": [], "subscriptions": [], }, "user_profile": { "first_name": "", "last_name": "", "name": "", "dietary_prefs": [], "fitness_goal": "", "budget_weekly": 0, "address": "", "city": "", "postal_code": "", "country": "", }, } def first_run_data() -> dict: """What a fresh install starts with: demo persona if LIFEOS_DEMO, else blank.""" return seed_data() if config.DEMO else empty_data() def seed_data() -> dict: """Demo persona: Awais, a busy student on a tight budget.""" return { "meals": [ {"date": _d(7), "dish": "Chicken karahi", "ingredients": ["chicken thighs", "tomatoes", "ginger", "green chili", "yogurt"]}, {"date": _d(6), "dish": "Spaghetti aglio e olio", "ingredients": ["spaghetti", "garlic", "olive oil", "chili flakes"]}, {"date": _d(5), "dish": "Egg fried rice", "ingredients": ["rice", "eggs", "frozen peas", "soy sauce", "green onion"]}, {"date": _d(4), "dish": "Spaghetti bolognese", "ingredients": ["spaghetti", "ground beef", "tomato sauce", "onion"]}, {"date": _d(3), "dish": "Chickpea curry", "ingredients": ["chickpeas", "coconut milk", "tomatoes", "curry powder", "rice"]}, {"date": _d(2), "dish": "Grilled cheese and tomato soup", "ingredients": ["bread", "cheddar", "canned tomato soup"]}, {"date": _d(1), "dish": "Chicken caesar wrap", "ingredients": ["chicken breast", "tortilla", "romaine", "caesar dressing", "parmesan"]}, ], "workouts": [ {"date": _d(6), "type": "run", "duration_min": 30}, {"date": _d(5), "type": "push (chest/triceps)", "duration_min": 50}, {"date": _d(3), "type": "pull (back/biceps)", "duration_min": 45}, {"date": _d(2), "type": "run", "duration_min": 25}, {"date": _d(1), "type": "legs", "duration_min": 55}, ], "calendar": [ {"id": _new_id(), "title": "CS lecture: Distributed Systems", "date": _week_day(0), "start": "10:00", "end": "12:00", "kind": "class"}, {"id": _new_id(), "title": "Cafe shift", "date": _week_day(1), "start": "16:00", "end": "21:00", "kind": "work"}, {"id": _new_id(), "title": "Stats tutorial", "date": _week_day(2), "start": "09:00", "end": "10:30", "kind": "class"}, {"id": _new_id(), "title": "Cafe shift", "date": _week_day(5), "start": "08:00", "end": "14:00", "kind": "work"}, ], "workout_schedule": {"days": ["mon", "wed", "sat"], "time": "07:00"}, "goals": [ { "id": _new_id(), "title": "Used car", "target_amount": 8000, "saved": 1200, "deadline": _months_out(14), "notes": "", } ], "finances": { "income_monthly": 1400, "monthly_payments": [ {"name": "Rent (shared)", "amount": 650, "due_day": 1}, {"name": "Phone plan", "amount": 45, "due_day": 15}, {"name": "Transit pass", "amount": 128, "due_day": 1}, ], "subscriptions": [ {"name": "Netflix", "cost": 16.99, "last_used": _d(2)}, {"name": "Spotify Premium", "cost": 11.99, "last_used": _d(0)}, {"name": "Crunchyroll", "cost": 11.99, "last_used": _d(54)}, {"name": "FitnessPal Pro", "cost": 19.99, "last_used": _d(88)}, {"name": "Disney+", "cost": 13.99, "last_used": _d(41)}, {"name": "iCloud 200GB", "cost": 3.99, "last_used": _d(0)}, {"name": "Adobe CC Student", "cost": 22.99, "last_used": _d(9)}, ], }, "user_profile": { "first_name": "Awais", "last_name": "Aziz", "name": "Awais Aziz", "dietary_prefs": ["halal", "high-protein", "no pork"], "fitness_goal": "build muscle, run a 10K in September", "budget_weekly": 80, "address": "", "city": "Toronto", "postal_code": "", "country": "Canada", }, } def load() -> dict: with _lock: first_run = not os.path.exists(MEMORY_PATH) if first_run: mem = first_run_data() os.makedirs(DATA_DIR, exist_ok=True) with open(MEMORY_PATH, "w", encoding="utf-8") as f: json.dump(mem, f, indent=2, ensure_ascii=False) return mem with open(MEMORY_PATH, "r", encoding="utf-8") as f: mem = json.load(f) # Backfill keys added by schema upgrades so old memory.json files keep # working. Defaults come from empty_data() — never inject demo persona data # into a real user's store. blank = empty_data() changed = False for key in ("calendar", "workout_schedule", "goals"): if key not in mem: mem[key] = blank[key] changed = True if "monthly_payments" not in mem.get("finances", {}): mem.setdefault("finances", {})["monthly_payments"] = blank["finances"]["monthly_payments"] changed = True profile = mem.setdefault("user_profile", {}) if "first_name" not in profile: # Upgrade legacy single-`name` profiles: first word -> first_name. parts = (profile.get("name") or "").split(" ", 1) profile["first_name"] = parts[0] profile.setdefault("last_name", parts[1] if len(parts) > 1 else "") changed = True profile.setdefault("last_name", "") for key in ("address", "city", "postal_code", "country"): if key not in profile: profile[key] = blank["user_profile"][key] changed = True full_name = f"{profile['first_name']} {profile['last_name']}".strip() if profile.get("name") != full_name: profile["name"] = full_name changed = True for g in mem.get("goals", []): if not g.get("id"): g["id"] = _new_id() changed = True if changed: save(mem) return mem def save(mem: dict) -> None: with _lock: os.makedirs(DATA_DIR, exist_ok=True) with open(MEMORY_PATH, "w", encoding="utf-8") as f: json.dump(mem, f, indent=2, ensure_ascii=False) def _write(mem: dict) -> dict: """Persist a full store to disk and return it (used by the reset helpers). Caller must not hold _lock — save() takes it.""" save(mem) return mem def reset_to_empty() -> dict: """Wipe the store back to a blank slate (real-user reset).""" return _write(empty_data()) def reset_to_seed() -> dict: """Reset the store to the demo persona (used by demo mode / reset button).""" return _write(seed_data()) def log_meal(dish: str, ingredients: list[str], when: str | None = None) -> dict: mem = load() mem["meals"].append({"date": when or date.today().isoformat(), "dish": dish, "ingredients": ingredients}) save(mem) return mem def log_workout(workout_type: str, duration_min: int, when: str | None = None) -> dict: mem = load() mem["workouts"].append({"date": when or date.today().isoformat(), "type": workout_type, "duration_min": int(duration_min)}) save(mem) return mem def set_subscriptions(subs: list[dict]) -> dict: """Replace detected subscriptions (from the Money feature).""" mem = load() mem["finances"]["subscriptions"] = subs save(mem) return mem def add_event(ev: dict) -> dict: """Add a calendar event. Fills in an id if missing. Returns full mem.""" mem = load() event = { "id": ev.get("id") or _new_id(), "title": ev.get("title", ""), "date": ev.get("date", date.today().isoformat()), "start": ev.get("start", "09:00"), "end": ev.get("end", "10:00"), "kind": ev.get("kind", "other"), } mem.setdefault("calendar", []).append(event) save(mem) return mem def delete_event(event_id: str) -> dict: mem = load() mem["calendar"] = [e for e in mem.get("calendar", []) if e.get("id") != event_id] save(mem) return mem def set_workout_schedule(days: list[str], time: str) -> dict: mem = load() mem["workout_schedule"] = {"days": [d.lower()[:3] for d in days], "time": time} save(mem) return mem def set_monthly_payments(payments: list[dict]) -> dict: mem = load() mem.setdefault("finances", {})["monthly_payments"] = [ {"name": p.get("name", ""), "amount": p.get("amount", 0), "due_day": int(p.get("due_day", 1))} for p in payments ] save(mem) return mem def upsert_goal(goal: dict) -> dict: """Insert or update a goal by id (new id assigned when missing/unknown).""" mem = load() goals = mem.setdefault("goals", []) gid = goal.get("id") merged = { "id": gid or _new_id(), "title": goal.get("title", ""), "target_amount": goal.get("target_amount", 0), "saved": goal.get("saved", 0), "deadline": goal.get("deadline", _months_out(12)), "notes": goal.get("notes", ""), } for i, g in enumerate(goals): if gid and g.get("id") == gid: goals[i] = {**g, **{k: v for k, v in goal.items() if v is not None}} break else: goals.append(merged) save(mem) return mem def set_profile(fields: dict) -> dict: """Merge fields into user_profile; income_monthly routes to finances.""" mem = load() fields = dict(fields) if "income_monthly" in fields: mem.setdefault("finances", {})["income_monthly"] = fields.pop("income_monthly") profile = mem.setdefault("user_profile", {}) profile.update(fields) profile["name"] = f"{profile.get('first_name', '')} {profile.get('last_name', '')}".strip() or profile.get("name", "") save(mem) return mem def delete_goal(goal_id: str) -> dict: mem = load() mem["goals"] = [g for g in mem.get("goals", []) if g.get("id") != goal_id] save(mem) return mem # Sections the frontend may replace wholesale; value = path into mem. _EDITABLE_SECTIONS = { "meals": ("meals",), "workouts": ("workouts",), "calendar": ("calendar",), "subscriptions": ("finances", "subscriptions"), "monthly_payments": ("finances", "monthly_payments"), } def set_section(section: str, items: list) -> dict: """Replace a whitelisted list section wholesale (manual memory editing).""" path = _EDITABLE_SECTIONS.get(section) if path is None: raise ValueError(f"unknown section: {section!r}") if not isinstance(items, list): raise ValueError("items must be a list") mem = load() target = mem for key in path[:-1]: target = target.setdefault(key, {}) target[path[-1]] = items save(mem) return mem def events_in_window(days: int = 7, mem: dict | None = None) -> list[dict]: """Calendar events from today through today+days, sorted.""" mem = mem or load() start = date.today().isoformat() end = (date.today() + timedelta(days=days)).isoformat() return sorted( (e for e in mem.get("calendar", []) if start <= e.get("date", "") <= end), key=lambda e: (e.get("date", ""), e.get("start", "")), ) def recent_meals(days: int = 7, mem: dict | None = None) -> list[dict]: mem = mem or load() cutoff = (date.today() - timedelta(days=days)).isoformat() return [m for m in mem["meals"] if m["date"] >= cutoff] def workouts_in_window(days: int = 14, mem: dict | None = None) -> list[dict]: mem = mem or load() cutoff = (date.today() - timedelta(days=days)).isoformat() return sorted((w for w in mem["workouts"] if w["date"] >= cutoff), key=lambda w: w["date"])