from __future__ import annotations # Acorn single-file build (auto-generated) # ===== llm ===== """Tiny client for a LOCAL, OpenAI-compatible model endpoint. Works unchanged with: * Ollama -> http://localhost:11434/v1 (default) * llama.cpp server-> http://localhost:8080/v1 (set OPENAI_BASE_URL) If no local model is reachable, every call returns ``None`` and the caller falls back to deterministic templates. That is what keeps Acorn "always runs" even on a fresh Hugging Face Space with no model installed. Nothing here ever talks to a cloud provider — the whole point of the project. """ import os try: import requests except Exception: # pragma: no cover - requests is in requirements requests = None # Default to Ollama's OpenAI-compatible endpoint. Override for llama.cpp. BASE_URL = os.environ.get("OPENAI_BASE_URL", "http://localhost:11434/v1").rstrip("/") MODEL = os.environ.get("ACORN_MODEL", "qwen2.5:7b-instruct") API_KEY = os.environ.get("OPENAI_API_KEY", "local") # local servers ignore this def available(timeout: float = 0.6) -> bool: """Best-effort check that a local model server is up. Never raises.""" if requests is None: return False try: r = requests.get( BASE_URL + "/models", headers={"Authorization": f"Bearer {API_KEY}"}, timeout=timeout, ) return r.ok except Exception: return False def chat(messages, temperature: float = 0.4, max_tokens: int = 400, timeout: float = 30.0): """Return assistant text, or ``None`` on any failure (caller falls back).""" if requests is None: return None try: r = requests.post( BASE_URL + "/chat/completions", headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}, json={ "model": MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False, }, timeout=timeout, ) r.raise_for_status() return r.json()["choices"][0]["message"]["content"].strip() except Exception: return None def info() -> dict: return {"base_url": BASE_URL, "model": MODEL, "available": available()} # ===== profile ===== """User profile + onboarding state, persisted as JSON on-device. This is half of Acorn's memory: the *declared* preferences from onboarding. The other half (learned baselines) lives in py. """ import json import os ACORN_HOME = os.environ.get("ACORN_HOME", os.path.expanduser("~/.acorn")) os.makedirs(ACORN_HOME, exist_ok=True) PROFILE_PATH = os.path.join(ACORN_HOME, "profile.json") # Onboarding maps directly to the user's original wishlist. GOAL_CHOICES = [ "Track where my time goes", "Block distracting apps on demand", "Remind me of things", "Keep my screen easy on the eyes", "Warn me about bloat / power hogs", "Manage my clipboard", "Organize my screenshots", "Suggest setup tweaks for my role", ] CONSENT_LEVELS = ["observe", "suggest", "act"] # silent -> nudge -> one-tap DEFAULT = { "name": "", "role": "Software / ML engineer", "goals": list(GOAL_CHOICES), "consent": "suggest", "screenshot_dir": os.path.expanduser("~/Desktop"), "quiet_hours_start": 21, # warm the screen after this hour "break_interval_min": 25, # force a breathing break after this much continuous use "break_duration_sec": 90, # how long the break overlay stays "break_strict": False, # strict = no skip until the break finishes "onboarded": False, } def profile_load() -> dict: try: with open(PROFILE_PATH, "r", encoding="utf-8") as f: data = json.load(f) merged = dict(DEFAULT) merged.update(data or {}) return merged except Exception: return dict(DEFAULT) def profile_save(profile: dict) -> dict: merged = dict(DEFAULT) merged.update(profile or {}) try: with open(PROFILE_PATH, "w", encoding="utf-8") as f: json.dump(merged, f, indent=2) except Exception: pass return merged # ===== store ===== """Local SQLite memory — the substrate that lets Acorn 'get smarter over time'. Stores telemetry samples, clipboard history, screenshot index, app-usage rollups, and emitted Pure stdlib (sqlite3). Lives under ~/.acorn. """ import hashlib import os import sqlite3 import threading import time ACORN_HOME = os.environ.get("ACORN_HOME", os.path.expanduser("~/.acorn")) os.makedirs(ACORN_HOME, exist_ok=True) DB_PATH = os.path.join(ACORN_HOME, "acorn.db") _lock = threading.Lock() def _conn(): c = sqlite3.connect(DB_PATH) c.row_factory = sqlite3.Row return c SCHEMA = """ CREATE TABLE IF NOT EXISTS samples ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, cpu REAL, mem_pct REAL, mem_used_gb REAL, swap_gb REAL, battery REAL, power_plugged INTEGER, top_proc TEXT, top_proc_gb REAL ); CREATE TABLE IF NOT EXISTS clips ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, kind TEXT, preview TEXT, hash TEXT UNIQUE, count INTEGER DEFAULT 1, pinned INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS shots ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, path TEXT UNIQUE, title TEXT, ocr TEXT, filed INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS appusage ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, app TEXT, seconds REAL, category TEXT ); CREATE TABLE IF NOT EXISTS nudges ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, module TEXT, severity TEXT, title TEXT, body TEXT, action_id TEXT, status TEXT DEFAULT 'new' ); """ def init_db(): with _lock, _conn() as c: c.executescript(SCHEMA) # ---- samples ------------------------------------------------------------- def record_sample(v: dict): with _lock, _conn() as c: c.execute( "INSERT INTO samples(ts,cpu,mem_pct,mem_used_gb,swap_gb,battery," "power_plugged,top_proc,top_proc_gb) VALUES(?,?,?,?,?,?,?,?,?)", (time.time(), v.get("cpu"), v.get("mem_pct"), v.get("mem_used_gb"), v.get("swap_gb"), v.get("battery"), int(bool(v.get("power_plugged"))), v.get("top_proc"), v.get("top_proc_gb")), ) def recent_samples(n: int = 60): with _lock, _conn() as c: rows = c.execute("SELECT * FROM samples ORDER BY id DESC LIMIT ?", (n,)).fetchall() return [dict(r) for r in rows] # ---- clipboard ----------------------------------------------------------- def upsert_clip(preview: str, kind: str = "text") -> int: h = hashlib.sha1(preview.encode("utf-8", "ignore")).hexdigest() with _lock, _conn() as c: row = c.execute("SELECT id,count FROM clips WHERE hash=?", (h,)).fetchone() if row: c.execute("UPDATE clips SET count=count+1, ts=? WHERE id=?", (time.time(), row["id"])) return row["count"] + 1 c.execute("INSERT INTO clips(ts,kind,preview,hash) VALUES(?,?,?,?)", (time.time(), kind, preview[:500], h)) return 1 def top_clips(n: int = 20): with _lock, _conn() as c: rows = c.execute( "SELECT * FROM clips ORDER BY pinned DESC, count DESC, ts DESC LIMIT ?", (n,)).fetchall() return [dict(r) for r in rows] def pin_clip(clip_id: int): with _lock, _conn() as c: c.execute("UPDATE clips SET pinned=1 WHERE id=?", (clip_id,)) # ---- screenshots --------------------------------------------------------- def add_shot(path: str, title: str = "", ocr: str = "") -> bool: with _lock, _conn() as c: try: c.execute("INSERT INTO shots(ts,path,title,ocr) VALUES(?,?,?,?)", (time.time(), path, title, ocr)) return True except sqlite3.IntegrityError: return False # already indexed def recent_shots(n: int = 20): with _lock, _conn() as c: rows = c.execute("SELECT * FROM shots ORDER BY ts DESC LIMIT ?", (n,)).fetchall() return [dict(r) for r in rows] def search_shots(q: str, n: int = 20): like = f"%{q}%" with _lock, _conn() as c: rows = c.execute( "SELECT * FROM shots WHERE title LIKE ? OR ocr LIKE ? " "ORDER BY ts DESC LIMIT ?", (like, like, n)).fetchall() return [dict(r) for r in rows] # ---- app usage ----------------------------------------------------------- def add_appusage(app: str, seconds: float, category: str = "other"): with _lock, _conn() as c: c.execute("INSERT INTO appusage(ts,app,seconds,category) VALUES(?,?,?,?)", (time.time(), app, seconds, category)) def usage_rollup(since_hours: float = 24): cutoff = time.time() - since_hours * 3600 with _lock, _conn() as c: rows = c.execute( "SELECT app, category, SUM(seconds) AS secs FROM appusage " "WHERE ts>=? GROUP BY app ORDER BY secs DESC", (cutoff,)).fetchall() return [dict(r) for r in rows] # ---- nudges -------------------------------------------------------------- def save_nudge(n: dict): with _lock, _conn() as c: cur = c.execute( "INSERT INTO nudges(ts,module,severity,title,body,action_id,status) " "VALUES(?,?,?,?,?,?, 'new')", (time.time(), n.get("module"), n.get("severity"), n.get("title"), n.get("body"), n.get("action_id"))) return cur.lastrowid def feed_recent(n: int = 15): """Recent insights/nudges — the async engine's delivery feed.""" with _lock, _conn() as c: rows = c.execute("SELECT * FROM nudges ORDER BY id DESC LIMIT ?", (n,)).fetchall() return [dict(r) for r in rows] def set_feed_status(nudge_id: int, status: str): with _lock, _conn() as c: c.execute("UPDATE nudges SET status=? WHERE id=?", (status, nudge_id)) def _today_midnight() -> float: n = time.localtime() return time.mktime((n.tm_year, n.tm_mon, n.tm_mday, 0, 0, 0, 0, 0, -1)) def day_metrics() -> dict: """Aggregate everything Acorn remembers about *today* — feeds the """ midnight = _today_midnight() with _lock, _conn() as c: usage = [dict(r) for r in c.execute( "SELECT app, category, SUM(seconds) AS secs FROM appusage " "WHERE ts>=? GROUP BY app ORDER BY secs DESC", (midnight,)).fetchall()] switches = c.execute( "SELECT COUNT(*) FROM appusage WHERE ts>=?", (midnight,)).fetchone()[0] shots = c.execute( "SELECT COUNT(*) FROM shots WHERE ts>=?", (midnight,)).fetchone()[0] clips = c.execute( "SELECT COUNT(*) FROM clips WHERE ts>=?", (midnight,)).fetchone()[0] top_clip = c.execute( "SELECT preview, count FROM clips ORDER BY count DESC LIMIT 1").fetchone() nudge_n = c.execute( "SELECT COUNT(*) FROM nudges WHERE ts>=?", (midnight,)).fetchone()[0] peak = c.execute( "SELECT MAX(cpu), MAX(mem_pct) FROM samples WHERE ts>=?", (midnight,)).fetchone() cats = {} for u in usage: cats[u.get("category", "other")] = cats.get(u.get("category", "other"), 0) + (u["secs"] or 0) return { "usage": usage, "switches": switches, "shots": shots, "clips": clips, "top_clip": dict(top_clip) if top_clip else None, "nudges": nudge_n, "peak_cpu": peak[0] if peak else None, "peak_mem": peak[1] if peak else None, "active_seconds": sum((u["secs"] or 0) for u in usage), "categories": cats, } def stats() -> dict: with _lock, _conn() as c: def one(q): return c.execute(q).fetchone()[0] return { "samples": one("SELECT COUNT(*) FROM samples"), "clips": one("SELECT COUNT(*) FROM clips"), "shots": one("SELECT COUNT(*) FROM shots"), "nudges": one("SELECT COUNT(*) FROM nudges"), } # ===== sensors ===== """Sensors — cheap, poll-based observation of the machine. Two runtimes, chosen by platform: * MacSensors -> real macOS signals (psutil + pbpaste + folder scan + osascript) * DemoSensors -> for the Hugging Face Space / Linux: real psutil where it works (container CPU/RAM are genuinely live!) + clearly-labelled sample data for the mac-only signals. This dual-runtime design is how one codebase satisfies "must be a Space" while still being a real local Mac app (see build plan §4.3). """ import os import platform import shutil import subprocess import time try: import psutil except Exception: # pragma: no cover psutil = None IS_MAC = platform.system() == "Darwin" # -------------------------------------------------------------------------- class BaseSensors: runtime = "base" demo = True # ---- system vitals (psutil works on Mac AND in a Linux Space) ---- def vitals(self) -> dict: if psutil is None: return {"cpu": 0, "mem_pct": 0, "mem_used_gb": 0, "swap_gb": 0, "battery": None, "power_plugged": True, "top_proc": "n/a", "top_proc_gb": 0} vm = psutil.virtual_memory() sw = psutil.swap_memory() cpu = psutil.cpu_percent(interval=0.2) bat, plugged = None, True try: b = psutil.sensors_battery() if b is not None: bat, plugged = round(b.percent, 0), b.power_plugged except Exception: pass top = self.top_processes(1) tp = top[0] if top else {"name": "n/a", "mem_gb": 0} return { "cpu": round(cpu, 1), "mem_pct": round(vm.percent, 1), "mem_used_gb": round(vm.used / 1e9, 1), "mem_total_gb": round(vm.total / 1e9, 1), "swap_gb": round(sw.used / 1e9, 2), "battery": bat, "power_plugged": plugged, "top_proc": tp["name"], "top_proc_gb": tp["mem_gb"], } def top_processes(self, n: int = 5): if psutil is None: return [] procs = [] for p in psutil.process_iter(["name", "memory_info", "cpu_percent"]): try: mi = p.info["memory_info"] procs.append({ "name": p.info["name"] or "?", "mem_gb": round((mi.rss if mi else 0) / 1e9, 2), "cpu": p.info.get("cpu_percent") or 0.0, }) except Exception: continue procs.sort(key=lambda x: x["mem_gb"], reverse=True) return procs[:n] # ---- the mac-only signals: overridden by each runtime ---- def disk_reclaimable(self): raise NotImplementedError def clipboard_peek(self): raise NotImplementedError def recent_screenshots(self, limit=10): raise NotImplementedError def display_state(self): raise NotImplementedError def frontmost_app(self): raise NotImplementedError # -------------------------------------------------------------------------- class MacSensors(BaseSensors): runtime = "mac" demo = False DEV_PATHS = [ ("Xcode DerivedData", "~/Library/Developer/Xcode/DerivedData"), ("User caches", "~/Library/Caches"), ("iOS DeviceSupport", "~/Library/Developer/Xcode/iOS DeviceSupport"), ("npm cache", "~/.npm"), ("Old Downloads", "~/Downloads"), ] def _du(self, path: str) -> float: p = os.path.expanduser(path) if not os.path.exists(p): return 0.0 try: out = subprocess.run(["du", "-sk", p], capture_output=True, text=True, timeout=8).stdout kb = float(out.split("\t")[0]) return round(kb / 1e6, 2) # GB except Exception: return 0.0 def disk_reclaimable(self): items = [{"label": lbl, "path": pth, "gb": self._du(pth)} for lbl, pth in self.DEV_PATHS] items = [i for i in items if i["gb"] > 0.05] items.sort(key=lambda x: x["gb"], reverse=True) return items def clipboard_peek(self): try: out = subprocess.run(["pbpaste"], capture_output=True, text=True, timeout=3).stdout return out.strip()[:500] except Exception: return "" def recent_screenshots(self, limit=10): d = os.path.expanduser(profile_load().get("screenshot_dir", "~/Desktop")) shots = [] try: for fn in os.listdir(d): low = fn.lower() if (low.endswith((".png", ".jpg", ".jpeg")) and ("screenshot" in low or "screen shot" in low)): fp = os.path.join(d, fn) shots.append({"path": fp, "name": fn, "mtime": os.path.getmtime(fp)}) except Exception: pass shots.sort(key=lambda x: x["mtime"], reverse=True) return shots[:limit] def display_state(self): # Best-effort; real Night Shift state needs CoreBrightness. Report appearance. appearance = "unknown" try: out = subprocess.run( ["osascript", "-e", 'tell application "System Events" to tell appearance preferences ' 'to return dark mode'], capture_output=True, text=True, timeout=3).stdout appearance = "dark" if "true" in out.lower() else "light" except Exception: pass return {"appearance": appearance, "hour": time.localtime().tm_hour} def frontmost_app(self): try: out = subprocess.run( ["osascript", "-e", 'tell application "System Events" to get name of first application ' 'process whose frontmost is true'], capture_output=True, text=True, timeout=3).stdout.strip() return out or "Unknown" except Exception: return "Unknown" # -------------------------------------------------------------------------- class DemoSensors(BaseSensors): """For the Space / non-mac. Real psutil vitals + labelled sample data.""" runtime = "demo" demo = True def disk_reclaimable(self): return [ {"label": "Xcode DerivedData", "path": "~/Library/Developer/Xcode/DerivedData", "gb": 8.7}, {"label": "User caches", "path": "~/Library/Caches", "gb": 3.9}, {"label": "Old Downloads", "path": "~/Downloads", "gb": 1.6}, ] def clipboard_peek(self): return "sk_live_demo_1234 (sample clipboard content)" def recent_screenshots(self, limit=10): now = time.time() sample = [ {"path": "~/Desktop/Screenshot 2026-06-08 at 10.14.png", "name": "Screenshot — Stripe webhook docs", "mtime": now - 600}, {"path": "~/Desktop/Screenshot 2026-06-08 at 09.02.png", "name": "Screenshot — stack trace KeyError", "mtime": now - 4200}, {"path": "~/Desktop/Screenshot 2026-06-07 at 18.20.png", "name": "Screenshot — Figma button states", "mtime": now - 60000}, ] return sample[:limit] def display_state(self): return {"appearance": "light", "hour": time.localtime().tm_hour} def frontmost_app(self): return "Visual Studio Code" def get_sensors() -> BaseSensors: return MacSensors() if IS_MAC else DemoSensors() # ===== nudges ===== """The Brain — turns signals + profile into nudges, a brief, and answers. The small local model is load-bearing but used for what small models are good at: ranking, classification, and warm phrasing. If no model is reachable we fall back to deterministic templates so Acorn ALWAYS runs. """ import time VOICE = ( "You are Acorn, a calm, brief macOS companion (a friendly squirrel). " "You run fully on-device and never send data to the cloud. " "Speak warmly, in 1-2 short sentences, never naggy. Be specific and kind." ) # ---- gather a snapshot of everything we know right now ------------------- def gather_signals(sensors, profile) -> dict: v = sensors.vitals() return { "vitals": v, "top_processes": sensors.top_processes(5), "reclaimable": sensors.disk_reclaimable(), "screenshots": sensors.recent_screenshots(8), "clips": top_clips(8), "usage": usage_rollup(24), "display": sensors.display_state(), "hour": time.localtime().tm_hour, "profile": profile, "runtime": sensors.runtime, } # ---- rule-based candidate nudges (deterministic, always available) ------- def rule_nudges(sig: dict) -> list: out = [] v = sig["vitals"] prof = sig["profile"] goals = set(prof.get("goals", [])) # Pulse: a memory-hungry top process tp = (sig["top_processes"] or [{}])[0] if tp and tp.get("mem_gb", 0) >= 2.0 and "Warn me about bloat / power hogs" in goals: out.append({ "module": "Pulse", "severity": "warn", "title": f"{tp['name']} is holding {tp['mem_gb']} GB", "body": f"{tp['name']} is your biggest memory user right now.", "action_id": "quit_app", "action_label": f"Quit {tp['name']}", "action_args": {"name": tp["name"]}, }) # Pulse: reclaimable disk reclaim = sig.get("reclaimable", []) total = round(sum(i.get("gb", 0) for i in reclaim), 1) if total >= 2 and "Warn me about bloat / power hogs" in goals: out.append({ "module": "Pulse", "severity": "info", "title": f"Reclaim {total} GB of developer junk", "body": "Caches, DerivedData and old downloads are piling up.", "action_id": "preview_cleanup", "action_label": "Preview cleanup", "action_args": {"items": reclaim}, }) # Pulse: on battery and draining hard if (v.get("power_plugged") is False and (v.get("cpu", 0) > 60) and "Warn me about bloat / power hogs" in goals): out.append({ "module": "Pulse", "severity": "warn", "title": "High CPU on battery", "body": f"CPU at {v['cpu']}% while unplugged — something is working hard.", "action_id": "deep_work_on", "action_label": "Calm things down", "action_args": {}, }) # Stash: a recurring paste worth pinning for clip in sig.get("clips", []): if clip.get("count", 0) >= 3 and not clip.get("pinned") \ and "Manage my clipboard" in goals: out.append({ "module": "Stash", "severity": "info", "title": "You keep pasting this", "body": f"Copied {clip['count']}×: “{(clip.get('preview') or '')[:40]}…” — pin it?", "action_id": "pin_clip", "action_label": "Pin it", "action_args": {"clip_id": clip["id"]}, }) break # Stash: screenshots piling up shots = sig.get("screenshots", []) if len(shots) >= 3 and "Organize my screenshots" in goals: out.append({ "module": "Stash", "severity": "info", "title": f"{len(shots)} screenshots on your Desktop", "body": "Want me to title and file them so they're searchable?", "action_id": "reveal_path", "action_label": "Show newest", "action_args": {"path": shots[0]["path"]}, }) # Comfort: evening eye care if sig["hour"] >= prof.get("quiet_hours_start", 21) \ and "Keep my screen easy on the eyes" in goals: out.append({ "module": "Comfort", "severity": "info", "title": "It's getting late", "body": "Warm and dim the screen so your eyes can wind down?", "action_id": "warm_screen", "action_label": "Warm my screen", "action_args": {}, }) # Trails: distraction check (from learned usage) usage = sig.get("usage", []) distract = sum(u["secs"] for u in usage if u.get("category") == "distraction") if distract > 1800 and "Block distracting apps on demand" in goals: out.append({ "module": "Trails", "severity": "info", "title": f"{int(distract//60)} min in distractions today", "body": "Want a Deep Work block for the next stretch?", "action_id": "deep_work_on", "action_label": "Start Deep Work", "action_args": {}, }) return out # ---- optional LLM re-phrasing of a single nudge -------------------------- def phrase_nudge(nudge: dict) -> dict: text = chat([ {"role": "system", "content": VOICE}, {"role": "user", "content": f"Rewrite this nudge as one warm sentence. Title: {nudge['title']}. " f"Detail: {nudge['body']}"}, ], max_tokens=60) if text: nudge = dict(nudge) nudge["body"] = text return nudge def nudges(sensors, profile, use_llm: bool = True) -> list: sig = gather_signals(sensors, profile) cands = rule_nudges(sig) for n in cands: save_nudge(n) if use_llm and available(): cands = [phrase_nudge(n) for n in cands[:5]] return cands # ---- the daily brief ----------------------------------------------------- def daily_brief(sensors, profile) -> str: sig = gather_signals(sensors, profile) v = sig["vitals"] total = round(sum(i.get("gb", 0) for i in sig.get("reclaimable", [])), 1) facts = ( f"CPU {v['cpu']}%, RAM {v['mem_pct']}% " f"({v['mem_used_gb']}/{v.get('mem_total_gb','?')} GB), " f"top process {v['top_proc']} ({v['top_proc_gb']} GB). " f"~{total} GB reclaimable. " f"{len(sig.get('screenshots', []))} recent screenshots, " f"{len(sig.get('clips', []))} clipboard items remembered. " f"Local hour {sig['hour']}." ) text = chat([ {"role": "system", "content": VOICE}, {"role": "user", "content": f"Write a 2-3 sentence morning-style brief for a {profile.get('role')}. " f"Be specific, warm, end with one gentle suggestion. Facts: {facts}"}, ], max_tokens=160) if text: return text # template fallback return ( f"🌰 Here's your Mac right now: CPU {v['cpu']}%, RAM {v['mem_pct']}%, " f"with {v['top_proc']} using the most memory ({v['top_proc_gb']} GB). " f"I can reclaim about {total} GB whenever you like. " + ("Late evening — say the word and I'll warm your screen." if sig['hour'] >= profile.get('quiet_hours_start', 21) else "Looking tidy. I'll keep watch.") ) # ---- ask-me-anything ----------------------------------------------------- def answer(question: str, sensors, profile) -> str: sig = gather_signals(sensors, profile) v = sig["vitals"] context = ( f"vitals={v}; top={sig['top_processes'][:3]}; " f"reclaimable={sig['reclaimable']}; clips={[c.get('preview') for c in sig['clips'][:5]]}; " f"screenshots={[s.get('name') for s in sig['screenshots'][:5]]}; hour={sig['hour']}" ) text = chat([ {"role": "system", "content": VOICE + " Answer ONLY from the provided local data. If unknown, say so plainly."}, {"role": "user", "content": f"Data: {context}\n\nQuestion: {question}"}, ], max_tokens=220) if text: return text return (f"Right now: CPU {v['cpu']}%, RAM {v['mem_pct']}%, top memory user " f"{v['top_proc']} ({v['top_proc_gb']} GB). " "(Install a local model via Ollama for fuller answers.)") # ---- 'people in your role configure X' ----------------------------------- DEV_CONFIG_TIPS = [ "Show all filename extensions in Finder", "Set a fast key-repeat rate (great for vim/editors)", "Save screenshots to a dedicated ~/Screenshots folder", "Auto-hide the Dock to reclaim screen space", "Install a window manager (Rectangle) and a launcher (Raycast)", "Enable tap-to-click and three-finger drag", "Turn on FileVault disk encryption", ] def config_suggestions(profile) -> str: tips = "; ".join(DEV_CONFIG_TIPS) text = chat([ {"role": "system", "content": VOICE}, {"role": "user", "content": f"For a {profile.get('role')}, pick the 5 most useful Mac setup tweaks " f"from this list and present them as a short friendly checklist: {tips}"}, ], max_tokens=220) if text: return text return "Recommended for your role:\n" + "\n".join(f"- [ ] {t}" for t in DEV_CONFIG_TIPS) # ===== journal ===== """Acorn's Almanac — the End-of-Day Journal. At day's end, the local model reads everything Acorn remembered today (app time, context switches, screenshots, clipboard, nudges, peak load, reclaimable space) and writes a warm narrative + takeaways. Saved as dated markdown under ~/.acorn/journal/. Only possible *because* the memory is local — that's the point. """ import os import time JOURNAL_DIR = os.path.join(ACORN_HOME, "journal") os.makedirs(JOURNAL_DIR, exist_ok=True) def _facts(sensors, profile): m = day_metrics() v = sensors.vitals() reclaim = round(sum(i.get("gb", 0) for i in sensors.disk_reclaimable()), 1) top = m["usage"][:4] active_min = int(m["active_seconds"] // 60) top_str = ", ".join(f"{u['app']} {int((u['secs'] or 0)//60)}m" for u in top) \ or "no apps tracked yet today" facts = ( f"active ~{active_min} min; app switches {m['switches']}; top apps {top_str}; " f"screenshots today {m['shots']}; clipboard items {m['clips']}; " f"nudges surfaced {m['nudges']}; peak CPU {m['peak_cpu']}; peak RAM {m['peak_mem']}; " f"~{reclaim} GB reclaimable now." ) return m, active_min, top_str, reclaim, facts def _template(date, active_min, top_str, m, reclaim) -> str: return ( f"Today your Mac was busy for about **{active_min} minutes** of tracked activity, " f"mostly across {top_str}. You switched apps {m['switches']} times, took {m['shots']} " f"screenshots, and Acorn kept {m['clips']} clipboard items for you. " f"Peak load touched CPU {m['peak_cpu']}% / RAM {m['peak_mem']}%. " f"There's about {reclaim} GB you could reclaim whenever you like.\n\n" f"**Takeaways**\n\n" f"- Most of your time went to {top_str.split(',')[0] if top_str else 'focused work'}.\n" f"- {m['switches']} context switches — batching similar work could cut that down.\n" f"- {reclaim} GB of reclaimable space is quietly piling up.\n\n" f"**For tomorrow:** start with your heaviest task while the machine is fresh — " f"and let me reclaim that {reclaim} GB so you begin with a tidy burrow. 🌰" ) def write_journal(sensors, profile) -> tuple[str, str]: """Generate today's journal, save it, return (markdown, path).""" m, active_min, top_str, reclaim, facts = _facts(sensors, profile) date = time.strftime("%Y-%m-%d") body = chat( [ {"role": "system", "content": VOICE}, {"role": "user", "content": f"Write an end-of-day journal entry for a {profile.get('role')} about how they used " f"their Mac today. 4-6 warm sentences, then a '**Takeaways**' list of 3 short bullets, " f"then one '**For tomorrow:**' suggestion. Be specific to the facts; do not invent " f"numbers. Facts: {facts}"}, ], max_tokens=420, ) if not body: body = _template(date, active_min, top_str, m, reclaim) md = f"# 🌰 Acorn Almanac — {date}\n\n_A diary of your Mac. Written on-device; never leaves it._\n\n{body}\n" path = os.path.join(JOURNAL_DIR, f"{date}.md") try: with open(path, "w", encoding="utf-8") as f: f.write(md) except Exception: pass return md, path def list_journals(): try: files = sorted((f for f in os.listdir(JOURNAL_DIR) if f.endswith(".md")), reverse=True) return [os.path.join(JOURNAL_DIR, f) for f in files] except Exception: return [] # ===== async_engine ===== """Async Insight Engine — the layer that makes Acorn a *companion*, not a dashboard. A background loop evaluates triggers (interval health sweep, heads-down/break rhythm, rule nudges), ranks candidates, and delivers the best one to: * an in-app feed (persisted via store), and * macOS Notification Center (osascript) — so it reaches you without opening the app. Guardrails keep it from nagging: consent level, quiet hours, an hourly nudge budget, per-title de-dupe, and a dismissed-set. All on-device; notifications make no network calls. """ import collections import platform import subprocess import threading import time IS_MAC = platform.system() == "Darwin" def notify(title: str, message: str) -> bool: """Post a macOS notification. No-op (returns False) off-mac.""" if not IS_MAC: return False safe = message.replace('"', "'") try: subprocess.run( ["osascript", "-e", f'display notification "{safe}" with title "{title}"'], timeout=5, capture_output=True) return True except Exception: return False class AsyncEngine: def __init__(self, sensors, get_profile, interval=60, break_minutes=50, max_per_hour=3): self.sensors = sensors self.get_profile = get_profile # callable -> current profile dict self.interval = interval # seconds between sweeps self.break_minutes = break_minutes self.max_per_hour = max_per_hour self._thread = None self._stop = threading.Event() self._emitted = collections.deque() # timestamps, for the hourly budget self._last_title = {} # title -> ts, for de-dupe self.dismissed = set() # titles the user dismissed -> back off self.active_since = time.time() # for heads-down detection self.log = [] # (time, title, delivery) for the UI # ---- lifecycle ---- def running(self) -> bool: return self._thread is not None and self._thread.is_alive() def start(self): if self.running(): return self._stop.clear() self.active_since = time.time() self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() def stop(self): self._stop.set() self._thread = None # ---- guardrails ---- def _quiet(self, prof) -> bool: h = time.localtime().tm_hour return h >= int(prof.get("quiet_hours_start", 21)) or h < 7 def _budget_ok(self) -> bool: now = time.time() while self._emitted and now - self._emitted[0] > 3600: self._emitted.popleft() return len(self._emitted) < self.max_per_hour # ---- candidate insights ---- def _candidates(self, prof): sig = gather_signals(self.sensors, prof) cands = rule_nudges(sig) # Heads-down / break rhythm (async-only signal). active_min = (time.time() - self.active_since) / 60 if active_min >= self.break_minutes and "Remind me of things" in prof.get("goals", []): cands.insert(0, { "module": "Trails", "severity": "info", "title": f"Heads-down {int(active_min)} min — take a breather?", "body": "Open the Breathe tab for a slow 4·7·8 — two minutes resets your focus.", "action_id": "warm_screen", "action_label": "Breathe", "action_args": {}, }) self.active_since = time.time() return cands # ---- one evaluation pass ---- def tick(self, force=False): prof = self.get_profile() consent = prof.get("consent", "suggest") delivered = [] for n in self._candidates(prof): title = n["title"] if title in self.dismissed: continue if not force: if time.time() - self._last_title.get(title, 0) < 1800: # 30-min de-dupe continue if not self._budget_ok(): break save_nudge(n) self._last_title[title] = time.time() self._emitted.append(time.time()) sent = False if consent in ("suggest", "act") and (force or not self._quiet(prof)): sent = notify("Acorn 🌰", title) self.log.append((time.strftime("%H:%M:%S"), title, "🔔 notified" if sent else "📋 feed")) delivered.append(n) if not force: # one good nudge per sweep is plenty break return delivered def dismiss(self, title: str): if title: self.dismissed.add(title) def _loop(self): while not self._stop.wait(self.interval): try: self.tick() except Exception: pass def status(self) -> str: if not self.running(): return "💤 Companion is asleep." return (f"🟢 Companion is watching · sweep every {self.interval}s · " f"break after {self.break_minutes}m · ≤{self.max_per_hour} nudges/hr" + ("" if IS_MAC else " · (notifications are mac-only; showing in feed)")) # ===== actions ===== """Actions — the one-tap macOS effects behind each nudge. Design rules (match the user's safety preferences): * Consent-gated: destructive actions require an explicit confirm. * Dry-run first: cleanup PREVIEWS sizes; only deletes when confirmed AND ACORN_ALLOW_DELETE=1 is set (belt-and-suspenders against accidents). * Safe on the Space: on non-mac, actions become labelled no-ops. Uses native macOS power moves: Shortcuts, Focus, osascript, defaults, pmset. """ import os import platform import shutil import subprocess IS_MAC = platform.system() == "Darwin" ALLOW_DELETE = os.environ.get("ACORN_ALLOW_DELETE") == "1" def _osa(script: str): if not IS_MAC: return True, "(demo) applied — this runs for real on your Mac." try: subprocess.run(["osascript", "-e", script], check=True, capture_output=True, text=True, timeout=6) return True, "Done." except Exception as e: return False, f"AppleScript failed: {e}" # ---- comfort ------------------------------------------------------------- def warm_screen(args=None, confirm=False): """Enable a warm, dimmed screen (Night Shift-style) for the evening.""" if not IS_MAC: return True, "🌙 (demo) On your Mac this warms + dims the screen for evening." # Toggling Night Shift programmatically needs CoreBrightness; we approximate # by enabling Dark Mode + lowering brightness via a user Shortcut hook. ok, _ = _osa( 'tell application "System Events" to tell appearance preferences ' 'to set dark mode to true') return ok, ("🌙 Warmed the screen (Dark Mode on; tip: bind brightness to a Shortcut)." if ok else "Couldn't reach System Events.") def dark_mode_toggle(args=None, confirm=False): if not IS_MAC: return True, "🌗 (demo) On your Mac this toggles Dark Mode." return _osa( 'tell application "System Events" to tell appearance preferences ' 'to set dark mode to not dark mode') # ---- focus --------------------------------------------------------------- def deep_work_on(args=None, confirm=False): """Turn on a Focus mode for deep work (best-effort via Shortcuts).""" if not IS_MAC: return True, "(demo) would enable a 'Deep Work' Focus and block distractors." # Most robust path: a user-created Shortcut named 'Deep Work'. return run_shortcut({"name": "Deep Work"}) def run_shortcut(args, confirm=False): name = (args or {}).get("name", "") if not name: return False, "No Shortcut name given." if not IS_MAC: return True, f"(demo) would run Shortcut: {name!r}" try: subprocess.run(["shortcuts", "run", name], check=True, capture_output=True, text=True, timeout=10) return True, f"▶️ Ran Shortcut: {name}" except Exception as e: return False, f"Couldn't run Shortcut {name!r} (create it in the Shortcuts app). {e}" # ---- pulse / cleanup ----------------------------------------------------- def preview_cleanup(args, confirm=False): """Dry-run: report how much could be reclaimed. Never deletes.""" items = (args or {}).get("items", []) total = sum(i.get("gb", 0) for i in items) lines = "\n".join(f" • {i['label']}: {i['gb']} GB" for i in items) return True, f"Dry-run — {round(total,1)} GB reclaimable:\n{lines}\n(Confirm to clean.)" def reclaim_space(args, confirm=False): """Delete cache contents — ONLY with confirm AND ACORN_ALLOW_DELETE=1.""" items = (args or {}).get("items", []) total = round(sum(i.get("gb", 0) for i in items), 1) if not confirm: return False, f"Needs confirmation to reclaim {total} GB." if not IS_MAC: return True, f"(demo) would reclaim ~{total} GB from caches/DerivedData." if not ALLOW_DELETE: return True, (f"Safe mode: would reclaim ~{total} GB. " "Set ACORN_ALLOW_DELETE=1 to enable real deletion.") freed = 0.0 for i in items: p = os.path.expanduser(i.get("path", "")) # Only ever clear *contents* of known cache dirs, never the dir itself. if "Caches" in p or "DerivedData" in p: try: for entry in os.listdir(p): fp = os.path.join(p, entry) shutil.rmtree(fp, ignore_errors=True) if os.path.isdir(fp) \ else os.remove(fp) freed += i.get("gb", 0) except Exception: continue return True, f"🧹 Reclaimed ~{round(freed,1)} GB." def quit_app(args, confirm=False): name = (args or {}).get("name", "") if not confirm: return False, f"Confirm to quit {name}." if not IS_MAC: return True, f"(demo) would quit {name}." return _osa(f'tell application "{name}" to quit') # ---- stash --------------------------------------------------------------- def reveal_path(args, confirm=False): path = os.path.expanduser((args or {}).get("path", "")) if not IS_MAC: return True, f"(demo) would reveal in Finder: {path}" try: subprocess.run(["open", "-R", path], check=True, timeout=5) return True, f"📁 Revealed {os.path.basename(path)} in Finder." except Exception as e: return False, str(e) # Registry: action_id -> callable(args, confirm) REGISTRY = { "warm_screen": warm_screen, "dark_mode_toggle": dark_mode_toggle, "deep_work_on": deep_work_on, "run_shortcut": run_shortcut, "preview_cleanup": preview_cleanup, "reclaim_space": reclaim_space, "quit_app": quit_app, "reveal_path": reveal_path, } def run(action_id: str, args=None, confirm=False): fn = REGISTRY.get(action_id) if not fn: return False, f"Unknown action: {action_id}" try: return fn(args or {}, confirm=confirm) except Exception as e: return False, f"Action error: {e}" # ===== spellbook ===== """Spellbook — the local model as a macOS automation author. You describe what you want ("tile my two coding windows", "hide everything but the front app") and the on-device model writes the AppleScript. Acorn never runs generated code blind: generate -> SHOW the script + plain-English explanation -> validate -> you click Run Safety model: * A denylist validator blocks dangerous patterns (shell-out, deletion, admin rights, keystroke injection, network) before anything can run. * A curated, pre-vetted LIBRARY guarantees the demo works even if the model is slow or wrong — the model handles novel requests. * Recipes prefer REVERSIBLE actions (hide / minimize / move) over close / quit, so you never lose unsaved work. All on-device. Generation uses the local model; nothing is sent to the cloud. Note: controlling other apps' windows uses System Events, which needs macOS Accessibility permission (granted once in System Settings → Privacy & Security). """ import os import platform import re import subprocess import time IS_MAC = platform.system() == "Darwin" SPELL_DIR = os.path.join(ACORN_HOME, "spells") os.makedirs(SPELL_DIR, exist_ok=True) # Reusable AppleScript prelude: screen size + frontmost process. _PRELUDE = '''tell application "Finder" to set _d to bounds of window of desktop set _sw to item 3 of _d set _sh to item 4 of _d tell application "System Events" to set _p to first application process whose frontmost is true''' # ---- curated, pre-vetted library ---------------------------------------- LIBRARY = [ { "name": "Tile front window — left half", "desc": "Snap the active window to the left half of the screen.", "script": _PRELUDE + ''' tell application "System Events" set position of window 1 of _p to {0, 25} set size of window 1 of _p to {_sw / 2, _sh - 25} end tell''', }, { "name": "Tile front window — right half", "desc": "Snap the active window to the right half of the screen.", "script": _PRELUDE + ''' tell application "System Events" set position of window 1 of _p to {_sw / 2, 25} set size of window 1 of _p to {_sw / 2, _sh - 25} end tell''', }, { "name": "Maximize front window", "desc": "Make the active window fill the screen (below the menu bar).", "script": _PRELUDE + ''' tell application "System Events" set position of window 1 of _p to {0, 25} set size of window 1 of _p to {_sw, _sh - 25} end tell''', }, { "name": "Center front window", "desc": "Center the active window at a comfortable reading size.", "script": _PRELUDE + ''' set _w to _sw * 0.66 set _h to (_sh - 25) * 0.85 tell application "System Events" set size of window 1 of _p to {_w, _h} set position of window 1 of _p to {(_sw - _w) / 2, 25 + ((_sh - 25) - _h) / 2} end tell''', }, { "name": "Full height (keep width)", "desc": "Stretch the active window to full height without changing its width.", "script": _PRELUDE + ''' tell application "System Events" set _pos to position of window 1 of _p set _sz to size of window 1 of _p set position of window 1 of _p to {item 1 of _pos, 25} set size of window 1 of _p to {item 1 of _sz, _sh - 25} end tell''', }, { "name": "Hide background apps", "desc": "Hide every app except the one in front (reversible — nothing is closed).", "script": '''tell application "System Events" set visible of (every application process whose frontmost is false and background only is false) to false end tell''', }, { "name": "Cascade all windows", "desc": "Fan the front window of each visible app into a tidy cascade.", "script": '''tell application "System Events" set _i to 0 repeat with _proc in (every application process whose visible is true and background only is false) try set position of window 1 of _proc to {40 + _i * 40, 50 + _i * 36} set _i to _i + 1 end try end repeat end tell''', }, { "name": "Move front window to 2nd display", "desc": "Nudge the active window onto a second display to the right.", "script": _PRELUDE + ''' tell application "System Events" set position of window 1 of _p to {_sw + 60, 60} end tell''', }, ] def library_names(): return [s["name"] for s in LIBRARY] def get_library(name): return next((s for s in LIBRARY if s["name"] == name), None) # ---- safety validator ---------------------------------------------------- # Patterns that must never appear in a script we run. _DENY = [ r"do\s+shell\s+script", r"\bsudo\b", r"administrator\s+privileges", r"\brm\s+-", r"\bdelete\b", r"\bmove\b.+\bto\s+trash", r"empty\s+trash", r"keystroke", r"key\s+code", r"\bcurl\b", r"\bwget\b", r"/bin/", r"/usr/bin", r'tell\s+application\s+"Terminal"', r"set\s+volume", r"shut\s*down", r"restart\b", r"log\s+out", r"\bpassword\b", r"system\s+attribute", ] def is_safe(script: str): low = (script or "").lower() for pat in _DENY: if re.search(pat, low): return False, f"blocked: matched unsafe pattern /{pat}/" if not low.strip(): return False, "empty script" return True, "ok" # ---- model generation ---------------------------------------------------- _SYS = ( "You are Acorn's automation author. Convert the user's request into a SINGLE " "macOS AppleScript suitable for `osascript`. Rules: " "(1) Output ONLY AppleScript — no markdown fences, no prose. " "(2) The FIRST line must be a comment '-- '. " "(3) Control windows with System Events; get screen size from " "`tell application \"Finder\" to get bounds of window of desktop`. " "(4) Prefer REVERSIBLE actions (hide, minimize, move, resize) over closing or " "quitting apps. " "(5) NEVER use: do shell script, administrator privileges, keystroke, delete, " "empty trash, set volume, shutdown/restart/logout, or anything touching files, " "the network, or passwords. Window and layout automation only." ) def generate(request: str, profile=None): """NL -> {script, explanation, source}. Falls back to the closest library spell.""" text = chat( [{"role": "system", "content": _SYS}, {"role": "user", "content": request}], temperature=0.2, max_tokens=320, ) if text: script = text.strip().strip("`") # pull the leading comment as the explanation expl = "" m = re.match(r"\s*--\s*(.+)", script) if m: expl = m.group(1).strip() ok, reason = is_safe(script) if ok: return {"script": script, "explanation": expl or "Generated automation.", "source": "model", "safe": True} return {"script": script, "explanation": f"⚠️ Generated, but {reason}. Not runnable.", "source": "model", "safe": False} # fallback: keyword-match a library spell so the feature still works offline spell = _closest_library(request) if spell: return {"script": spell["script"], "explanation": f"(no local model — used a library spell: {spell['name']})", "source": "library", "safe": True} return {"script": "", "explanation": "No local model available and no close library match. Try Ollama, or pick a library spell.", "source": "none", "safe": False} def _closest_library(request: str): r = (request or "").lower() keys = { "left": 0, "right": 1, "max": 2, "full screen": 2, "center": 3, "centre": 3, "height": 4, "hide": 5, "minimi": 5, "cascade": 6, "stack": 6, "display": 7, "monitor": 7, "second screen": 7, } for k, idx in keys.items(): if k in r: return LIBRARY[idx] return None # ---- run ----------------------------------------------------------------- def spell_run(script: str, confirm: bool = False): ok, reason = is_safe(script) if not ok: return False, f"🛑 Refused — {reason}" if not confirm: return True, "Preview only — click Run to execute." if not IS_MAC: return True, "(demo) On your Mac this AppleScript would run via osascript." try: path = os.path.join(SPELL_DIR, "_run.applescript") with open(path, "w", encoding="utf-8") as f: f.write(script) out = subprocess.run(["osascript", path], capture_output=True, text=True, timeout=12) if out.returncode == 0: return True, "✨ Done." + (f" {out.stdout.strip()}" if out.stdout.strip() else "") return False, f"AppleScript error: {out.stderr.strip()[:200]} "\ "(System Events automations need Accessibility permission)." except Exception as e: return False, f"Run error: {e}" def save_spell(name: str, script: str): ok, reason = is_safe(script) if not ok: return False, f"Won't save — {reason}" safe_name = re.sub(r"[^A-Za-z0-9_-]+", "_", (name or "spell").strip()) or "spell" path = os.path.join(SPELL_DIR, f"{safe_name}.applescript") try: with open(path, "w", encoding="utf-8") as f: f.write(script) return True, f"🪄 Saved spell: `{path}`" except Exception as e: return False, f"Save error: {e}" def list_spells(): try: return [os.path.join(SPELL_DIR, f) for f in sorted(os.listdir(SPELL_DIR)) if f.endswith(".applescript") and not f.startswith("_")] except Exception: return [] # ===== app ===== """Acorn — your Mac's quiet woodland companion. Build Small Hackathon (Gradio × Hugging Face) · Backyard AI track. Local-first: runs entirely on-device, no cloud APIs. Run locally on your Mac for the full thing: pip install -r requirements.txt python app.py (Optional, for the AI brief/nudges) install Ollama and `ollama pull qwen2.5:7b-instruct`. On a Hugging Face Space (Linux) it runs in demo mode automatically. """ import os import threading import time import gradio as gr # ---- singletons ---------------------------------------------------------- init_db() SENSORS = get_sensors() PROFILE = profile_load() ENGINE = AsyncEngine(SENSORS, lambda: PROFILE, interval=60, break_minutes=50, max_per_hour=3) ACCENT = "#7a9e6e" # ---- small render helpers ------------------------------------------------ def meter(label: str, pct, detail: str = "") -> str: pct = max(0.0, min(100.0, float(pct or 0))) color = ACCENT if pct < 70 else ("#d9a441" if pct < 88 else "#c4694e") return ( f'
' f'{label}{detail}
' f'
' ) def render_vitals_html() -> str: v = SENSORS.vitals() record_sample(v) bat = "—" if v.get("battery") is None else f"{int(v['battery'])}%" plug = "plugged in" if v.get("power_plugged") else "on battery" html = [f'
'] html.append(meter("CPU", v["cpu"], f"{v['cpu']}%")) html.append(meter("Memory", v["mem_pct"], f"{v['mem_used_gb']} / {v.get('mem_total_gb','?')} GB")) html.append( f'
🔋 Battery{bat} · {plug}
') html.append( f'
🧠 Top memory' f'{v["top_proc"]} · {v["top_proc_gb"]} GB
') if SENSORS.demo: html.append('
Live CPU/RAM are real (this container). ' 'Mac-only signals show sample data — full power runs on your Mac.
') html.append("
") return "\n".join(html) def render_top_rows(): return [[p["name"], p["mem_gb"], p["cpu"]] for p in SENSORS.top_processes(8)] _RECLAIM_CACHE = {"items": []} def render_reclaim(): items = SENSORS.disk_reclaimable() _RECLAIM_CACHE["items"] = items total = round(sum(i["gb"] for i in items), 1) lines = "\n".join(f"- **{i['label']}** — {i['gb']} GB \n `{i['path']}`" for i in items) return f"### ♻️ ~{total} GB reclaimable\n{lines or '_Nothing notable right now._'}" def render_clip_rows(): return [["📌" if c.get("pinned") else "", (c.get("preview") or "")[:60], c.get("count", 1)] for c in top_clips(15)] def render_shot_rows(): shots = SENSORS.recent_screenshots(12) for s in shots: add_shot(s["path"], title=s.get("name", ""), ocr=s.get("name", "")) return [[s.get("name", os.path.basename(s["path"])), s["path"]] for s in shots] CATEGORY = { "Code": "coding", "Visual Studio Code": "coding", "Xcode": "coding", "iTerm": "coding", "Terminal": "coding", "PyCharm": "coding", "Slack": "comms", "Mail": "comms", "Messages": "comms", "Safari": "browsing", "Google Chrome": "browsing", "Arc": "browsing", "Twitter": "distraction", "X": "distraction", "YouTube": "distraction", } def categorize(app: str) -> str: return CATEGORY.get(app, "other") def render_usage_rows(): rows = usage_rollup(24) if not rows: # seed a tiny demo picture so the tab isn't empty on first run demo = [("Visual Studio Code", "coding", 14400), ("Google Chrome", "browsing", 5400), ("Slack", "comms", 3600), ("Twitter", "distraction", 2100)] return [[a, c, f"{s//60} min"] for a, c, s in demo] return [[r["app"], r.get("category", "other"), f"{int(r['secs']//60)} min"] for r in rows] # ---- callbacks ----------------------------------------------------------- def save_onboarding(name, role, goals, consent, quiet_hour, break_interval=25, strict=False): global PROFILE PROFILE = profile_save({ "name": name, "role": role, "goals": goals or [], "consent": consent, "quiet_hours_start": int(quiet_hour or 21), "break_interval_min": int(break_interval or 25), "break_strict": bool(strict), "onboarded": True, }) mode = "strict" if strict else "skippable" return (f"🌰 Saved. Acorn will {consent} for {len(goals or [])} habits · " f"{mode} breathing break every {int(break_interval or 25)} min.") def refresh_brief(): brief = daily_brief(SENSORS, PROFILE) cand = nudges(SENSORS, PROFILE) if not cand: cards = "_All calm — nothing needs you right now._" else: cards = "\n\n".join( f'
' f'{n["title"]}
{n["body"]}
' f'→ {n["action_label"]}
' for n in cand) titles = [n["title"] for n in cand] return brief, cards, gr.update(choices=titles, value=(titles[0] if titles else None)), cand def apply_nudge(title, cand): if not title or not cand: return "Pick a suggestion first." n = next((x for x in cand if x["title"] == title), None) if not n: return "That suggestion is gone — refresh the brief." aid, args = n["action_id"], n.get("action_args", {}) if aid == "pin_clip": pin_clip(args.get("clip_id")) return "📌 Pinned that clip." ok, msg = run(aid, args, confirm=True) return ("✅ " if ok else "⚠️ ") + msg def pulse_refresh(): return render_vitals_html(), render_top_rows(), render_reclaim() def do_preview_cleanup(): ok, msg = run("preview_cleanup", {"items": _RECLAIM_CACHE["items"]}) return msg def do_reclaim(): ok, msg = run("reclaim_space", {"items": _RECLAIM_CACHE["items"]}, confirm=True) return ("✅ " if ok else "⚠️ ") + msg def capture_clipboard(): text = SENSORS.clipboard_peek() if text: count = upsert_clip(text) return render_clip_rows(), f"Stashed (seen {count}×)." return render_clip_rows(), "Clipboard was empty." def search_screenshots(q): if not q: return render_shot_rows() render_shot_rows() # ensure indexed hits = search_shots(q) return [[h.get("title") or os.path.basename(h["path"]), h["path"]] for h in hits] def deep_work(on: bool): ok, msg = run("deep_work_on" if on else "run_shortcut", {"name": "Deep Work Off"}, confirm=True) return ("✅ " if ok else "⚠️ ") + msg def comfort_action(which): ok, msg = run(which, {}, confirm=True) return ("✅ " if ok else "⚠️ ") + msg def get_config_tips(): return config_suggestions(PROFILE) def chat_respond(message, history): history = history or [] reply = answer(message, SENSORS, PROFILE) history = history + [{"role": "user", "content": message}, {"role": "assistant", "content": reply}] return history, "" # ---- Almanac (end-of-day journal) ---- def gen_journal(): md, path = write_journal(SENSORS, PROFILE) return md, f"_Saved on-device: `{path}`_" # ---- Companion (async engine) ---- def render_engine_log(): if not ENGINE.log: return "_No insights delivered yet. Start the companion or run a check._" rows = list(reversed(ENGINE.log[-12:])) return "\n".join(f"- `{t}` · {how} · {title}" for (t, title, how) in rows) def engine_start(): ENGINE.start() return ENGINE.status(), render_engine_log() def engine_stop(): ENGINE.stop() return ENGINE.status(), render_engine_log() def engine_check(): ENGINE.tick(force=True) return ENGINE.status(), render_engine_log() # ---- Spellbook (model -> AppleScript automations) ---- def show_library_spell(name): sp = get_library(name) return sp["script"] if sp else "" def run_library_spell(name): sp = get_library(name) if not sp: return "Pick a spell first." ok, msg = spell_run(sp["script"], confirm=True) return msg def conjure(request): res = generate(request, PROFILE) note = ("✅ looks safe" if res.get("safe") else "⚠️ flagged by the validator") return res["script"], f"**{res['explanation']}** · _{res['source']}_ · {note}", res["script"] def run_generated(script): ok, msg = spell_run(script or "", confirm=True) return msg def save_generated(name, script): ok, msg = save_spell(name, script or "") return msg # ---- Forced breathing break (scheduled full-window overlay) ---- WORK_STATE = {"last_break": time.time(), "active": False, "started": 0.0} def break_tick(): """Runs on a Timer. Decides whether the break overlay should show, and updates it.""" now = time.time() interval = PROFILE.get("break_interval_min", 25) * 60 dur = PROFILE.get("break_duration_sec", 90) strict = PROFILE.get("break_strict", False) if not WORK_STATE["active"]: if now - WORK_STATE["last_break"] >= interval: WORK_STATE["active"] = True WORK_STATE["started"] = now else: return (gr.update(visible=False), "", gr.update(visible=True), gr.update(visible=True)) remaining = int(dur - (time.time() - WORK_STATE["started"])) if remaining <= 0: WORK_STATE["active"] = False WORK_STATE["last_break"] = time.time() return (gr.update(visible=False), "", gr.update(visible=True), gr.update(visible=True)) show = not strict msg = (f"Resume in {remaining}s — settle in." if strict else f"{remaining}s of calm · or skip whenever you're ready") return (gr.update(visible=True), msg, gr.update(visible=show), gr.update(visible=show)) def break_skip(): WORK_STATE["active"] = False WORK_STATE["last_break"] = time.time() return gr.update(visible=False) def break_postpone(): WORK_STATE["active"] = False WORK_STATE["last_break"] = time.time() - max(0, PROFILE.get("break_interval_min", 25) - 5) * 60 return gr.update(visible=False) def break_now(): WORK_STATE["active"] = True WORK_STATE["started"] = time.time() return gr.update(visible=True) # ---- UI ------------------------------------------------------------------ CUSTOM_CSS = r""" /* Acorn — woodland theme (Off-Brand / Custom UI badge) */ .gradio-container { max-width: 920px !important; margin: 0 auto !important; } .acorn-head { display:flex; align-items:center; gap:14px; padding:8px 4px 2px; } .acorn-head h1 { margin:0; font-size:1.7rem; color:#3f4a36; letter-spacing:.5px; } .acorn-head p { margin:0; color:#7a8470; font-size:.9rem; } .acorn-logo { font-size:2.4rem; filter:drop-shadow(0 2px 3px rgba(90,110,70,.3)); } .acorn-card { background:linear-gradient(180deg,#fbf9f3,#f3efe4); border:1px solid #e3ddcc; border-radius:16px; padding:16px 18px; } .acorn-meter { margin:10px 0; } .acorn-meter-label { display:flex; justify-content:space-between; font-size:.85rem; color:#5b6450; } .acorn-meter-label span { color:#9aa28c; } .acorn-bar { height:9px; background:#e6e0d0; border-radius:6px; overflow:hidden; margin-top:5px; } .acorn-fill { height:100%; border-radius:6px; transition:width .6s cubic-bezier(.2,.7,.3,1); } .acorn-row { display:flex; justify-content:space-between; padding:6px 2px; border-top:1px dashed #e3ddcc; font-size:.9rem; color:#5b6450; margin-top:6px; } .acorn-note { margin-top:10px; font-size:.78rem; color:#a08a5c; background:#faf4e3; border-radius:10px; padding:8px 10px; } .acorn-nudge { background:#fbf9f3; border:1px solid #e3ddcc; border-left:4px solid #7a9e6e; border-radius:12px; padding:12px 14px; margin:8px 0; } .acorn-nudge span { color:#6b7360; font-size:.9rem; } .acorn-nudge i { color:#7a9e6e; font-size:.82rem; } .acorn-warn { border-left-color:#d9a441; } .acorn-info { border-left-color:#7a9e6e; } button.primary, .primary button { border-radius:12px !important; } /* Breathing guide */ .abr-wrap{display:flex;flex-direction:column;align-items:center;gap:22px;padding:26px 0 10px} .abr-stage{position:relative;display:grid;place-items:center;width:340px;height:340px} .abr-stage>*{grid-area:1/1} .abr-ring{width:272px;height:272px;border-radius:50%;border:1px solid rgba(196,105,78,.18);animation:abr-ring 19s cubic-bezier(.37,0,.63,1) infinite} .abr-orb{width:220px;height:220px;border-radius:50%;background:radial-gradient(circle at 50% 45%, rgba(196,105,78,.30), rgba(217,164,65,.10) 42%, rgba(243,239,228,0) 72%);border:1px solid rgba(196,105,78,.28);box-shadow:0 0 80px rgba(217,164,65,.13);animation:abr-breathe 19s cubic-bezier(.37,0,.63,1) infinite} .abr-num{height:52px;overflow:hidden;line-height:52px} .abr-strip{animation:abr-count 19s steps(19,end) infinite} .abr-strip div{height:52px;font-size:32px;font-weight:300;color:#b07a52;text-align:center} @keyframes abr-count{to{transform:translateY(-988px)}} @keyframes abr-breathe{0%{transform:scale(.56)}21%{transform:scale(1)}58%{transform:scale(1)}100%{transform:scale(.56)}} @keyframes abr-ring{0%{transform:scale(.6);opacity:.22}21%{transform:scale(1.05);opacity:.5}58%{transform:scale(1.05);opacity:.5}100%{transform:scale(.6);opacity:.22}} .abr-phase{position:relative;height:1.5em;width:300px;text-align:center} .abr-phase span{position:absolute;inset:0;letter-spacing:.42em;text-transform:uppercase;font-size:1.02rem;color:#6f7a5e;opacity:0;animation:19s linear infinite} .abr-in{animation-name:abr-lin}.abr-hold{animation-name:abr-lhold}.abr-out{animation-name:abr-lout} @keyframes abr-lin{0%{opacity:1}19%{opacity:1}23%{opacity:0}100%{opacity:0}} @keyframes abr-lhold{0%{opacity:0}21%{opacity:0}25%{opacity:1}56%{opacity:1}60%{opacity:0}100%{opacity:0}} @keyframes abr-lout{0%{opacity:0}58%{opacity:0}62%{opacity:1}98%{opacity:1}100%{opacity:0}} .abr-cap{color:#a9b09c;font-size:.78rem;letter-spacing:.04em} @media (prefers-reduced-motion:reduce){.abr-orb,.abr-ring,.abr-strip,.abr-phase span{animation:none}.abr-orb{transform:scale(.85)}.abr-hold{opacity:1}.abr-strip{transform:none}} /* Full-window forced-break overlay */ #acorn-break{position:fixed;inset:0;z-index:9999;background:rgba(244,240,230,.97); align-items:center;justify-content:center;gap:6px;padding:40px} #acorn-break .acorn-break-title h3{color:#5b6450;font-weight:500;text-align:center} """ THEME = gr.themes.Soft(primary_hue="green") # A calm, CSS-only 4-7-8 breathing guide with a phase label + countdown # (no JS, so it renders inside gr.HTML). The number strip steps once per second. BREATHE_HTML = """
4
3
2
1
7
6
5
4
3
2
1
8
7
6
5
4
3
2
1
Breathe inHoldBreathe out
4 · 7 · 8 — follow the circle and the count
""" def build_app() -> gr.Blocks: # Inject CSS as a ") gr.HTML( '
' '

Acorn

your Mac\'s quiet woodland companion · ' '100% on-device

') model = info() runtime_label = ("🖥️ Mac (full power)" if not SENSORS.demo else "🧪 Demo mode (deploy/Linux)") model_label = (f"🦙 model: {model['model']} (live)" if model["available"] else "💤 no local model — using built-in rules (works fine)") gr.Markdown(f"**Runtime:** {runtime_label} · **{model_label}**") with gr.Tabs(): # ---- Brief ------------------------------------------------- with gr.Tab("🌰 Brief"): brief_md = gr.Markdown("_Press refresh for your brief._") brief_btn = gr.Button("Refresh brief & nudges", variant="primary") nudge_html = gr.HTML() with gr.Row(): nudge_pick = gr.Dropdown(label="Apply a suggestion", choices=[]) nudge_apply = gr.Button("Do it") nudge_result = gr.Markdown() nudge_state = gr.State([]) brief_btn.click(refresh_brief, None, [brief_md, nudge_html, nudge_pick, nudge_state], api_name="refresh_brief") nudge_apply.click(apply_nudge, [nudge_pick, nudge_state], nudge_result) # ---- Pulse ------------------------------------------------- with gr.Tab("🫀 Pulse"): gr.Markdown("System health & power.") pulse_btn = gr.Button("Refresh vitals", variant="primary") vitals_html = gr.HTML() gr.Markdown("**Top processes**") top_df = gr.Dataframe(headers=["Process", "Mem (GB)", "CPU %"], interactive=False, wrap=True) reclaim_md = gr.Markdown() with gr.Row(): prev_btn = gr.Button("Preview cleanup") reclaim_btn = gr.Button("Reclaim space (confirm)", variant="stop") reclaim_out = gr.Markdown() pulse_btn.click(pulse_refresh, None, [vitals_html, top_df, reclaim_md], api_name="pulse") prev_btn.click(do_preview_cleanup, None, reclaim_out) reclaim_btn.click(do_reclaim, None, reclaim_out) # ---- Stash ------------------------------------------------- with gr.Tab("🐿️ Stash"): gr.Markdown("Clipboard memory + screenshot inbox.") with gr.Row(): cap_btn = gr.Button("Capture current clipboard", variant="primary") cap_msg = gr.Markdown() clip_df = gr.Dataframe(headers=["", "Clip", "×"], interactive=False, wrap=True) gr.Markdown("**Screenshots** — searchable by their text/title") with gr.Row(): shot_q = gr.Textbox(label="Find a screenshot", placeholder="e.g. Stripe") shot_btn = gr.Button("Search") shot_df = gr.Dataframe(headers=["Screenshot", "Path"], interactive=False, wrap=True) cap_btn.click(capture_clipboard, None, [clip_df, cap_msg], api_name="capture_clipboard") shot_btn.click(search_screenshots, shot_q, shot_df) # ---- Trails ------------------------------------------------ with gr.Tab("🧭 Trails"): gr.Markdown("Where your time goes · focus on demand.") trails_btn = gr.Button("Refresh time map", variant="primary") usage_df = gr.Dataframe(headers=["App", "Category", "Time"], interactive=False, wrap=True) with gr.Row(): dw_on = gr.Button("Start Deep Work") dw_off = gr.Button("End Deep Work") dw_msg = gr.Markdown() trails_btn.click(lambda: render_usage_rows(), None, usage_df, api_name="trails") dw_on.click(lambda: deep_work(True), None, dw_msg, api_name="deep_work_on") dw_off.click(lambda: deep_work(False), None, dw_msg, api_name="deep_work_off") # ---- Comfort ----------------------------------------------- with gr.Tab("☀️ Comfort"): gr.Markdown("Eye comfort + setup tips for your role.") with gr.Row(): warm_btn = gr.Button("Warm my screen", variant="primary") dark_btn = gr.Button("Toggle Dark Mode") comfort_msg = gr.Markdown() gr.Markdown("**People in your role configure…**") tips_btn = gr.Button("Suggest setup tweaks") tips_md = gr.Markdown() warm_btn.click(lambda: comfort_action("warm_screen"), None, comfort_msg, api_name="warm_screen") dark_btn.click(lambda: comfort_action("dark_mode_toggle"), None, comfort_msg, api_name="dark_mode") tips_btn.click(get_config_tips, None, tips_md) # ---- Breathe ---------------------------------------------- with gr.Tab("🌬️ Breathe"): gr.Markdown("A slow **4·7·8** breath — in for 4, hold for 7, out for 8. Acorn also pops " "this up **full-screen on a schedule** (interval set in Setup) so you " "actually pause — skippable, or strict if you want no escape.") gr.HTML(BREATHE_HTML) try_break = gr.Button("Try a forced break now (preview)") # ---- Ask --------------------------------------------------- with gr.Tab("💬 Ask Acorn"): gr.Markdown("Ask anything about your Mac — answered from local data only.") chatbot = gr.Chatbot(height=320) msg = gr.Textbox(placeholder="what's eating my memory?", show_label=False) msg.submit(chat_respond, [msg, chatbot], [chatbot, msg], api_name="ask") # ---- Almanac (end-of-day journal) -------------------------- with gr.Tab("📔 Almanac"): gr.Markdown("Your **end-of-day journal** — the model writes a diary of your Mac " "from today's local memory. Written on-device; never leaves it.") alm_btn = gr.Button("Write today's journal", variant="primary") alm_md = gr.Markdown() alm_path = gr.Markdown() alm_btn.click(gen_journal, None, [alm_md, alm_path], api_name="journal") # ---- Companion (async insight engine) ---------------------- with gr.Tab("🔔 Companion"): gr.Markdown("Acorn watching in the **background**, surfacing the right nudge at the " "right time — delivered as macOS notifications + this feed.") comp_status = gr.Markdown(ENGINE.status()) with gr.Row(): comp_start = gr.Button("Start companion", variant="primary") comp_stop = gr.Button("Stop") comp_check = gr.Button("Run a check now") gr.Markdown("**Delivered insights**") comp_feed = gr.Markdown(render_engine_log()) comp_start.click(engine_start, None, [comp_status, comp_feed], api_name="companion_start") comp_stop.click(engine_stop, None, [comp_status, comp_feed], api_name="companion_stop") comp_check.click(engine_check, None, [comp_status, comp_feed], api_name="companion_check") # ---- Spellbook (model -> AppleScript automations) ---------- with gr.Tab("🪄 Spellbook"): gr.Markdown("The local model writes **macOS automations** for you. You always see " "the script and approve it before anything runs.") gr.Markdown("**Library — pre-vetted spells** (safe, reversible)") with gr.Row(): spell_pick = gr.Dropdown(library_names(), label="Pick a spell") spell_show = gr.Button("Show script") spell_run = gr.Button("Run spell", variant="primary") lib_code = gr.Code(label="AppleScript") lib_msg = gr.Markdown() spell_show.click(show_library_spell, spell_pick, lib_code) spell_run.click(run_library_spell, spell_pick, lib_msg, api_name="run_spell") gr.Markdown("**Conjure — describe an automation in plain English**") conj_in = gr.Textbox(label="e.g. tile my two front windows side by side") conj_btn = gr.Button("Conjure automation", variant="primary") gen_code = gr.Code(label="Generated AppleScript (review before running)") gen_expl = gr.Markdown() gen_state = gr.State("") with gr.Row(): gen_run = gr.Button("Run it", variant="stop") gen_name = gr.Textbox(label="Save as", scale=2) gen_save = gr.Button("Save spell") gen_msg = gr.Markdown() conj_btn.click(conjure, conj_in, [gen_code, gen_expl, gen_state], api_name="conjure") gen_run.click(run_generated, gen_state, gen_msg, api_name="run_generated") gen_save.click(save_generated, [gen_name, gen_state], gen_msg) gr.Markdown("_Safety: a validator blocks shell-out, deletion, keystrokes, volume/power, " "and anything destructive before it can run. Window automations need macOS " "Accessibility permission (granted once in System Settings)._") # ---- Settings ---------------------------------------------- with gr.Tab("⚙️ Setup"): gr.Markdown("### Onboarding — tell Acorn what to learn") name_in = gr.Textbox(label="Your name (optional)", value=PROFILE.get("name", "")) role_in = gr.Textbox(label="Your role", value=PROFILE.get("role", "")) goals_in = gr.CheckboxGroup(GOAL_CHOICES, label="What should I learn?", value=PROFILE.get("goals", [])) consent_in = gr.Radio(CONSENT_LEVELS, label="How forward should I be?", value=PROFILE.get("consent", "suggest"), info="observe = silent · suggest = nudge · act = one-tap") quiet_in = gr.Slider(18, 23, step=1, label="Warm my screen after (hour)", value=PROFILE.get("quiet_hours_start", 21)) gr.Markdown("**Breathing breaks**") interval_in = gr.Slider(1, 90, step=1, label="Force a breathing break every (minutes)", value=PROFILE.get("break_interval_min", 25), info="A full-screen 4·7·8 break appears after this much continuous use.") strict_in = gr.Checkbox(label="Strict mode — no skip until the break finishes", value=PROFILE.get("break_strict", False)) save_btn = gr.Button("Save", variant="primary") save_msg = gr.Markdown() save_btn.click(save_onboarding, [name_in, role_in, goals_in, consent_in, quiet_in, interval_in, strict_in], save_msg, api_name="save_setup") gr.Markdown("_Privacy: everything is stored in `~/.acorn` on this machine. " "Acorn makes no network calls._") # ---- Forced break overlay — covers the window, driven by a Timer ---- with gr.Column(visible=False, elem_id="acorn-break") as break_overlay: gr.Markdown("### 🌬️ Time for a breath", elem_classes="acorn-break-title") gr.HTML(BREATHE_HTML) break_remaining = gr.Markdown("") with gr.Row(): break_postpone_btn = gr.Button("Postpone 5 min") break_skip_btn = gr.Button("Skip", variant="primary") break_ticker = gr.Timer(2) break_ticker.tick(break_tick, None, [break_overlay, break_remaining, break_skip_btn, break_postpone_btn]) break_skip_btn.click(break_skip, None, break_overlay) break_postpone_btn.click(break_postpone, None, break_overlay) try_break.click(break_now, None, break_overlay) return demo # ---- optional background sampler (real usage learning) ------------------- def start_sampler(interval=15): def loop(): last_app, last_t = None, time.time() while True: try: v = SENSORS.vitals() record_sample(v) app = SENSORS.frontmost_app() now = time.time() if last_app: add_appusage(last_app, now - last_t, categorize(last_app)) last_app, last_t = app, now except Exception: pass time.sleep(interval) t = threading.Thread(target=loop, daemon=True) t.start() return t # Module-level app for Hugging Face Spaces (Spaces looks for `demo`). demo = build_app() if __name__ == "__main__": if os.environ.get("ACORN_NO_SAMPLER") != "1": start_sampler() demo.queue().launch(css=CUSTOM_CSS, theme=THEME, inbrowser=os.environ.get("SPACE_ID") is None)