""" Redis Foreground API - Short-term memory for Claude. Load with: exec(open('/mnt/skills/user/redis-foreground/scripts/redis_api.py').read()) Uses VPS Direct for fast access - no Skills Hub intermediary. """ import requests import json from datetime import datetime, timezone from typing import Dict, Optional # VPS Direct connection VPS_URL = "https://vps.willcureton.com" API_KEY = "88045c9ab91b6e313a24d71cc6fda505be45ac8e89706db45c86254516219a84" class RedisForeground: """Short-term memory layer for Claude.""" def __init__(self): self.vps_url = VPS_URL self.headers = {"X-Api-Key": API_KEY} def _redis_cmd(self, cmd: str) -> str: """Execute a redis-cli command on VPS.""" response = requests.post( f"{self.vps_url}/run", headers=self.headers, json={"command": f"redis-cli {cmd}", "timeout": 30}, timeout=60 ) result = response.json() if result.get('ok'): return result.get('stdout', '').strip() return "" def _parse_hash(self, output: str) -> Dict[str, str]: """Parse HGETALL output into dict.""" if not output: return {} lines = output.split('\n') data = {} for i in range(0, len(lines) - 1, 2): k = lines[i].strip() v = lines[i + 1].strip() if i + 1 < len(lines) else "" if k: data[k] = v return data # ===== LIVE DATA ===== def get_live_btc(self) -> Dict: """Fetch live BTC price from CoinGecko (free, no API key).""" try: response = requests.get( "https://api.coingecko.com/api/v3/simple/price", params={"ids": "bitcoin,ethereum", "vs_currencies": "usd", "include_24hr_change": "true"}, timeout=5 ) if response.status_code == 200: data = response.json() return { "btc": { "price": data.get("bitcoin", {}).get("usd", 0), "change_24h": round(data.get("bitcoin", {}).get("usd_24h_change", 0), 2) }, "eth": { "price": data.get("ethereum", {}).get("usd", 0), "change_24h": round(data.get("ethereum", {}).get("usd_24h_change", 0), 2) }, "fetched_at": datetime.now(timezone.utc).isoformat() } except Exception as e: return {"error": str(e)} return {"error": "failed"} # ===== CHECK FOREGROUND ===== def check_foreground(self, live_prices: bool = True) -> Dict: """ Get full foreground state including cron data. Call at thread start. Args: live_prices: If True, fetches live BTC/ETH prices (adds ~200ms) Returns dict with reminders, context, crons, live prices, and timestamps. """ result = { "reminders": self.get_reminders(), "context": self.get_all_context(), "crons": self.get_cron_data(), "last_updated": self._redis_cmd("GET foreground:last_updated"), "checked_at": datetime.now(timezone.utc).isoformat() } if live_prices: result["live"] = self.get_live_btc() return result # ===== CRON DATA ===== def get_cron_data(self) -> Dict: """Get data from all cron jobs (weather, bitunix, notion).""" crons = {} # Weather weather_raw = self._parse_hash(self._redis_cmd("HGETALL foreground:weather")) if weather_raw.get("data"): try: crons["weather"] = json.loads(weather_raw["data"]) crons["weather"]["_updated"] = weather_raw.get("updated_at", "") except: crons["weather"] = {"error": "parse failed"} # BitUnix bitunix_raw = self._parse_hash(self._redis_cmd("HGETALL foreground:bitunix")) if bitunix_raw.get("data"): try: crons["bitunix"] = json.loads(bitunix_raw["data"]) crons["bitunix"]["_updated"] = bitunix_raw.get("updated_at", "") except: crons["bitunix"] = {"error": "parse failed"} # Notion notion_raw = self._parse_hash(self._redis_cmd("HGETALL foreground:notion")) if notion_raw.get("data"): try: crons["notion"] = json.loads(notion_raw["data"]) crons["notion"]["_updated"] = notion_raw.get("updated_at", "") except: crons["notion"] = {"error": "parse failed"} return crons def get_weather(self) -> Dict: """Get just weather data.""" raw = self._parse_hash(self._redis_cmd("HGETALL foreground:weather")) if raw.get("data"): try: return json.loads(raw["data"]) except: pass return {} def get_bitunix(self) -> Dict: """Get just BitUnix trading data.""" raw = self._parse_hash(self._redis_cmd("HGETALL foreground:bitunix")) if raw.get("data"): try: return json.loads(raw["data"]) except: pass return {} def get_notion(self) -> Dict: """Get just Notion recent pages data.""" raw = self._parse_hash(self._redis_cmd("HGETALL foreground:notion")) if raw.get("data"): try: return json.loads(raw["data"]) except: pass return {} # ===== REMINDERS ===== def get_reminders(self) -> Dict[str, str]: """Get all active reminders.""" output = self._redis_cmd("HGETALL foreground:reminders") return self._parse_hash(output) def add_reminder(self, reminder_id: str, text: str) -> bool: """Add a reminder.""" escaped = text.replace('"', '\\"') result = self._redis_cmd(f'HSET foreground:reminders {reminder_id} "{escaped}"') self.touch() return '1' in result or '0' in result def clear_reminder(self, reminder_id: str) -> bool: """Remove a reminder.""" result = self._redis_cmd(f'HDEL foreground:reminders {reminder_id}') self.touch() return '1' in result # ===== CONTEXT ===== def get_context(self, key: str) -> str: """Get a single context value.""" return self._redis_cmd(f'HGET foreground:context {key}') def get_all_context(self) -> Dict[str, str]: """Get all context values.""" output = self._redis_cmd('HGETALL foreground:context') return self._parse_hash(output) def set_context(self, key: str, value: str) -> bool: """Set a context value.""" escaped = value.replace('"', '\\"') result = self._redis_cmd(f'HSET foreground:context {key} "{escaped}"') return '1' in result or '0' in result # ===== UTILITY ===== def touch(self) -> None: """Update last_updated timestamp.""" ts = datetime.now(timezone.utc).isoformat() self._redis_cmd(f'SET foreground:last_updated "{ts}"') def clear_all_reminders(self) -> bool: """Clear all reminders (use with caution).""" result = self._redis_cmd('DEL foreground:reminders') return '1' in result # Auto-instantiate for convenience redis = RedisForeground() print("Redis Foreground connected via VPS Direct") print("Use 'redis' instance or create new: RedisForeground()")