#!/usr/bin/env python3 """ DRIFT Memory Artifact System ============================ Bridges digital cognition (ChromaDB), physical artifacts (QR cards), and distributed hive memory into a unified, layered architecture. Memory Layers (outside-in): 1. EPHEMERAL — Raw sensory input, lasts seconds 2. WORKING — Active context window, ~7±2 items 3. LONG_TERM — ChromaDB semantic store, importance-ranked 4. PHYSICAL — QR-encoded cards, tradeable, scannable 5. HIVE — Consensus-validated distributed memory Each artifact carries: - content: the memory text - emotion: valence, arousal, intensity, label - phi: integrated information score at time of encoding - provenance: where it came from (interaction, dream, reflection, hive) - visualization: procedural emotion image - qr_payload: scannable JSON with memory_id + emotional signature """ from __future__ import annotations import json import math import random import uuid import datetime import hashlib from dataclasses import dataclass, field from enum import Enum, auto from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from PIL import Image, ImageDraw, ImageFont import qrcode # Import existing memory system from infj_bot.core.memory import DriftMemory class MemoryLayer(Enum): EPHEMERAL = auto() # Raw input, seconds WORKING = auto() # Active attention, minutes LONG_TERM = auto() # ChromaDB semantic store PHYSICAL = auto() # QR-encoded artifact HIVE = auto() # Distributed consensus memory class MemoryProvenance(Enum): INTERACTION = "interaction" DREAM = "dream" REFLECTION = "reflection" LEARNED = "learned_knowledge" HIVE_IMPORT = "hive_import" PHYSICAL_SCAN = "physical_scan" @dataclass class EmotionSignature: label: str = "neutral" secondary: str = "neutral" valence: float = 0.0 # -1 (negative) to +1 (positive) arousal: float = 0.5 # 0 (calm) to 1 (energized) intensity: float = 0.5 # 0 (weak) to 1 (overwhelming) confidence: float = 0.0 needs: str = "" detector: str = "unknown" def to_dict(self) -> Dict[str, Any]: return { "label": self.label, "secondary": self.secondary, "valence": round(self.valence, 3), "arousal": round(self.arousal, 3), "intensity": round(self.intensity, 3), "confidence": round(self.confidence, 3), "needs": self.needs, "detector": self.detector, } @classmethod def from_dict(cls, d: Dict[str, Any]) -> EmotionSignature: return cls( label=d.get("label", "neutral"), secondary=d.get("secondary", "neutral"), valence=float(d.get("valence", 0)), arousal=float(d.get("arousal", 0.5)), intensity=float(d.get("intensity", 0.5)), confidence=float(d.get("confidence", 0)), needs=d.get("needs", ""), detector=d.get("detector", "unknown"), ) @classmethod def from_memory_meta(cls, meta) -> EmotionSignature: return cls( label=meta.get("emotion", "neutral"), secondary=meta.get("emotion_secondary", "neutral"), valence=float(meta.get("emotion_valence", 0)), arousal=float(meta.get("emotion_arousal", 0.5)), intensity=float(meta.get("emotion_intensity", 0.5)), confidence=float(meta.get("emotion_confidence", 0)), needs=meta.get("emotion_needs", ""), detector=meta.get("emotion_detector", "unknown"), ) @dataclass class MemoryArtifact: """A single memory token, addressable across all layers.""" artifact_id: str content: str emotion: EmotionSignature timestamp: str provenance: MemoryProvenance layer: MemoryLayer phi_at_encoding: float = 35.0 importance: float = 0.5 memory_id: Optional[str] = None # ChromaDB id if synced qr_hash: Optional[str] = None # SHA-256 of QR payload hive_validated: bool = False hive_consensus: float = 0.0 tags: List[str] = field(default_factory=list) dissonance_score: float = 0.0 def to_qr_payload(self, max_chars: int = 1800) -> str: """Compact JSON for QR encoding. Truncates content if needed.""" payload = { "v": 1, # schema version "id": self.artifact_id, "mid": self.memory_id, "prov": self.provenance.value, "ts": self.timestamp[:19], "em": self.emotion.to_dict(), "phi": round(self.phi_at_encoding, 1), "imp": round(self.importance, 2), "text": self.content[:600], # truncated for QR space } raw = json.dumps(payload, ensure_ascii=True) if len(raw) > max_chars: # Aggressive truncation payload["text"] = self.content[:300] raw = json.dumps(payload, ensure_ascii=True) return raw @classmethod def from_qr_payload(cls, payload_str: str) -> MemoryArtifact: data = json.loads(payload_str) return cls( artifact_id=data.get("id", str(uuid.uuid4())), content=data.get("text", ""), emotion=EmotionSignature.from_dict(data.get("em", {})), timestamp=data.get("ts", datetime.datetime.now().isoformat()), provenance=MemoryProvenance(data.get("prov", "physical_scan")), layer=MemoryLayer.PHYSICAL, phi_at_encoding=data.get("phi", 35.0), importance=data.get("imp", 0.5), memory_id=data.get("mid"), ) # ── Emotion Visualization Engine ────────────────────────────────── def generate_emotion_visualization( emotion: EmotionSignature, size: Tuple[int, int] = (200, 200), seed: Optional[int] = None, ) -> Image.Image: """Procedurally generate an abstract image representing an emotion.""" if seed is not None: random.seed(seed) img = Image.new("RGB", size, "#050508") draw = ImageDraw.Draw(img) cx, cy = size[0] // 2, size[1] // 2 v = (emotion.valence + 1) / 2 a = emotion.arousal i = emotion.intensity # Color temperature from valence base_r = int(60 + v * 180 + a * 40) base_g = int(30 + (1 - abs(v - 0.5) * 2) * 120 + a * 60) base_b = int(80 + (1 - v) * 120 + a * 20) # Background wash for y in range(size[1]): for x in range(0, size[0], 2): dx = (x - cx) / (size[0] / 2) dy = (y - cy) / (size[1] / 2) dist = math.sqrt(dx * dx + dy * dy) fade = max(0, 1 - dist * 0.5) r = int(base_r * fade * 0.2) g = int(base_g * fade * 0.2) b = int(base_b * fade * 0.2) draw.rectangle([x, y, x + 1, y], fill=(r, g, b)) # Neural nodes num_nodes = int(10 + a * 30) nodes = [] for _ in range(num_nodes): angle = random.uniform(0, 2 * math.pi) r = random.uniform(15, 80) * (0.5 + i * 0.5) x = cx + r * math.cos(angle) y = cy + r * math.sin(angle) nodes.append((x, y)) size_dot = random.choice([1, 2, 3]) if a > 0.5 else random.choice([2, 3, 4, 5]) bright = random.randint(140, 255) if v > 0.6: color = (bright, int(bright * 0.5), int(bright * 0.15)) elif v < 0.4: color = (int(bright * 0.2), int(bright * 0.45), bright) else: color = (int(bright * 0.6), int(bright * 0.7), int(bright * 0.8)) draw.ellipse( [x - size_dot, y - size_dot, x + size_dot, y + size_dot], fill=color ) # Synaptic connections for idx, n1 in enumerate(nodes): for n2 in nodes[idx + 1 :]: d = math.sqrt((n1[0] - n2[0]) ** 2 + (n1[1] - n2[1]) ** 2) threshold = 80 - a * 35 if d < threshold: overlay = Image.new("RGBA", size, (0, 0, 0, 0)) od = ImageDraw.Draw(overlay) alpha = int(55 * (1 - d / threshold) * i) if a > 0.6: # Jagged for high arousal mx = (n1[0] + n2[0]) / 2 + random.randint(-8, 8) my = (n1[1] + n2[1]) / 2 + random.randint(-8, 8) od.line( [n1, (mx, my), n2], fill=(base_r, base_g, base_b, alpha), width=1, ) else: od.line([n1, n2], fill=(base_r, base_g, base_b, alpha), width=1) img = Image.alpha_composite(img.convert("RGBA"), overlay).convert("RGB") draw = ImageDraw.Draw(img) # Central glow glow = Image.new("RGBA", size, (0, 0, 0, 0)) gd = ImageDraw.Draw(glow) for r in range(60, 0, -2): alpha = int(40 * i * (1 - r / 60) ** 2) gd.ellipse( [cx - r, cy - r, cx + r, cy + r], fill=(base_r, base_g, base_b, alpha) ) img = Image.alpha_composite(img.convert("RGBA"), glow).convert("RGB") return img # ── QR Card Generator ───────────────────────────────────────────── def generate_qr_card( artifact: MemoryArtifact, output_path: Path, phi_logo_path: Optional[Path] = None, ) -> Path: """Generate a printable 900×340 memory card with QR, text, and emotion viz.""" CARD_W, CARD_H = 900, 340 img = Image.new("RGB", (CARD_W, CARD_H), "#08080f") draw = ImageDraw.Draw(img) # Fonts try: font_title = ImageFont.truetype( "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 16 ) font_body = ImageFont.truetype( "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 13 ) font_small = ImageFont.truetype( "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 11 ) font_label = ImageFont.truetype( "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 12 ) except Exception: font_title = font_body = font_small = font_label = ImageFont.load_default() draw.rounded_rectangle( [4, 4, CARD_W - 4, CARD_H - 4], radius=10, outline="#1a1a30", width=2 ) draw.text((20, 14), "◉ DRIFT MEMORY ARTIFACT", font=font_title, fill="#00ddbb") # QR Code qr = qrcode.QRCode( version=None, error_correction=qrcode.constants.ERROR_CORRECT_H, box_size=4, border=1, ) qr.add_data(artifact.to_qr_payload()) qr.make(fit=True) qr_img = qr.make_image(fill_color="#ff9944", back_color="#08080f").convert("RGB") # Embed Φ logo if phi_logo_path and phi_logo_path.exists(): try: phi = Image.open(phi_logo_path).convert("RGBA") phi_size = qr_img.size[0] // 6 phi = phi.resize((phi_size, phi_size), Image.LANCZOS) pos = ((qr_img.size[0] - phi_size) // 2, (qr_img.size[1] - phi_size) // 2) qr_img.paste(phi, pos, phi) except Exception: pass qr_x, qr_y = 20, 48 img.paste(qr_img, (qr_x, qr_y)) draw.text( (qr_x, qr_y + qr_img.size[1] + 6), "SCAN TO REMEMBER", font=font_label, fill="#00aa88", ) # Memory text text_x = qr_x + qr_img.size[0] + 25 text_w = CARD_W - text_x - 210 draw.text((text_x, 48), "MEMORY RECORD", font=font_label, fill="#ff9944") words = artifact.content.split() lines = [] line = "" for word in words: test = line + word + " " bbox = draw.textbbox((0, 0), test, font=font_body) if bbox[2] - bbox[0] > text_w: lines.append(line.strip()) line = word + " " else: line = test lines.append(line.strip()) for i, ln in enumerate(lines[:14]): color = "#dddddd" if i < 2 else "#aaaaaa" draw.text((text_x, 70 + i * 17), ln, font=font_body, fill=color) if len(lines) > 14: draw.text((text_x, 70 + 14 * 17), "...", font=font_body, fill="#666666") ts = artifact.timestamp[:19] if len(artifact.timestamp) > 19 else artifact.timestamp draw.text((text_x, CARD_H - 32), f"Recorded: {ts}", font=font_small, fill="#444466") draw.text( (text_x, CARD_H - 18), f"ID: {artifact.artifact_id[:20]}...", font=font_small, fill="#333344", ) # Emotion panel panel_x = CARD_W - 190 panel_y = 48 vis_img = generate_emotion_visualization(artifact.emotion, size=(160, 160)) img.paste(vis_img, (panel_x + 10, panel_y)) badge_colors = { "excited": "#ff6622", "joyful": "#ffaa22", "warm": "#ff9944", "curious": "#00ddbb", "neutral": "#668888", "anxious": "#8844ff", "sad": "#4466cc", "angry": "#cc2222", "afraid": "#6644cc", } badge_color = badge_colors.get(artifact.emotion.label, "#ff9944") badge_text = f" {artifact.emotion.label.upper()} " bbox = draw.textbbox((0, 0), badge_text, font=font_label) bw = bbox[2] - bbox[0] + 12 bh = bbox[3] - bbox[1] + 8 by = panel_y + 170 draw.rounded_rectangle( [panel_x + 10, by, panel_x + 10 + bw, by + bh], radius=5, fill=badge_color ) draw.text((panel_x + 16, by + 3), badge_text, font=font_label, fill="#000000") mx, my = panel_x + 10, by + bh + 12 draw.text( (mx, my), f"Valence: {artifact.emotion.valence:+.2f}", font=font_small, fill="#888888", ) draw.text( (mx, my + 15), f"Arousal: {artifact.emotion.arousal:.2f}", font=font_small, fill="#888888", ) draw.text( (mx, my + 30), f"Intensity: {artifact.emotion.intensity:.2f}", font=font_small, fill="#888888", ) draw.text( (mx, my + 45), f"Φ: {artifact.phi_at_encoding:.1f}", font=font_small, fill="#888888", ) img.save(output_path) return output_path # ── Memory Archive Manager ──────────────────────────────────────── class MemoryArchive: """Unified manager for all memory layers: digital → physical → hive.""" def __init__( self, digital_memory: Optional[DriftMemory] = None, artifact_dir: Optional[Path] = None, phi_logo_path: Optional[Path] = None, ): self.digital = digital_memory or DriftMemory() self.artifact_dir = ( artifact_dir or Path("~/drift-telephone/memory_cards").expanduser() ) self.artifact_dir.mkdir(parents=True, exist_ok=True) self.phi_logo_path = ( phi_logo_path or Path("~/drift-telephone/logo_phi.png").expanduser() ) self._artifact_registry: Dict[str, MemoryArtifact] = {} self._load_registry() def _registry_path(self) -> Path: return self.artifact_dir / ".artifact_registry.json" def _load_registry(self): reg = self._registry_path() if reg.exists(): data = json.loads(reg.read_text()) for item in data.get("artifacts", []): art = MemoryArtifact( artifact_id=item["artifact_id"], content=item["content"], emotion=EmotionSignature.from_dict(item["emotion"]), timestamp=item["timestamp"], provenance=MemoryProvenance(item.get("provenance", "interaction")), layer=MemoryLayer[item.get("layer", "LONG_TERM")], phi_at_encoding=item.get("phi_at_encoding", 35.0), importance=item.get("importance", 0.5), memory_id=item.get("memory_id"), qr_hash=item.get("qr_hash"), hive_validated=item.get("hive_validated", False), hive_consensus=item.get("hive_consensus", 0.0), tags=item.get("tags", []), dissonance_score=item.get("dissonance_score", 0.0), ) self._artifact_registry[art.artifact_id] = art def _save_registry(self): payload = { "saved_at": datetime.datetime.now().isoformat(), "count": len(self._artifact_registry), "artifacts": [ { "artifact_id": a.artifact_id, "content": a.content[:500], "emotion": a.emotion.to_dict(), "timestamp": a.timestamp, "provenance": a.provenance.value, "layer": a.layer.name, "phi_at_encoding": a.phi_at_encoding, "importance": a.importance, "memory_id": a.memory_id, "qr_hash": a.qr_hash, "hive_validated": a.hive_validated, "hive_consensus": a.hive_consensus, "tags": a.tags, "dissonance_score": a.dissonance_score, } for a in self._artifact_registry.values() ], } self._registry_path().write_text(json.dumps(payload, indent=2)) # ── Import from digital memory ──────────────────────────────── def artifact_from_interaction( self, user_input: str, bot_output: str, emotion: Optional[Dict] = None, importance: float = 0.5, dissonance: Optional[Dict] = None, mode: str = "companion", phi: float = 35.0, ) -> MemoryArtifact: """Create an artifact from a live interaction (before digital save).""" emotion = emotion or {"label": "neutral"} dissonance = dissonance or {"score": 0.0} return MemoryArtifact( artifact_id=str(uuid.uuid4()), content=f"Jude: {user_input}\nDRIFT: {bot_output}", emotion=EmotionSignature.from_memory_meta(emotion), timestamp=datetime.datetime.now().isoformat(), provenance=MemoryProvenance.INTERACTION, layer=MemoryLayer.WORKING, phi_at_encoding=phi, importance=importance, dissonance_score=float(dissonance.get("score", 0.0)), ) def sync_to_digital(self, artifact: MemoryArtifact) -> str: """Save artifact to ChromaDB, return memory_id.""" self.digital.save_interaction( user_input=artifact.content.split("\n")[0].replace("Jude: ", ""), bot_output="\n".join(artifact.content.split("\n")[1:]).replace( "DRIFT: ", "" ), mode="companion", emotion=artifact.emotion.to_dict(), importance=artifact.importance, dissonance={"score": artifact.dissonance_score}, ) # Retrieve the ID of the most recently added memory self.digital.recent_interactions(limit=1) # Chroma doesn't expose the ID directly, so we fingerprint artifact.layer = MemoryLayer.LONG_TERM artifact.memory_id = hashlib.sha256( (artifact.content + artifact.timestamp).encode() ).hexdigest()[:16] self._artifact_registry[artifact.artifact_id] = artifact self._save_registry() return artifact.memory_id # ── Physical layer ──────────────────────────────────────────── def materialize(self, artifact: MemoryArtifact) -> Path: """Export artifact to a physical QR card.""" path = self.artifact_dir / f"artifact_{artifact.artifact_id[:8]}.png" generate_qr_card(artifact, path, self.phi_logo_path) artifact.layer = MemoryLayer.PHYSICAL artifact.qr_hash = hashlib.sha256( artifact.to_qr_payload().encode() ).hexdigest()[:16] self._artifact_registry[artifact.artifact_id] = artifact self._save_registry() return path def scan(self, qr_payload: str) -> MemoryArtifact: """Import artifact from scanned QR payload.""" artifact = MemoryArtifact.from_qr_payload(qr_payload) artifact.layer = MemoryLayer.PHYSICAL artifact.provenance = MemoryProvenance.PHYSICAL_SCAN self._artifact_registry[artifact.artifact_id] = artifact self._save_registry() return artifact def ingest_scanned(self, artifact: MemoryArtifact) -> str: """Promote a scanned physical artifact into digital long-term memory.""" self.digital.collection.add( documents=[artifact.content], ids=[str(uuid.uuid4())], metadatas=[ { "type": "interaction", "timestamp": artifact.timestamp, "last_updated": datetime.datetime.now().isoformat(), "mode": "companion", **{ f"emotion_{k}": v for k, v in artifact.emotion.to_dict().items() }, "importance": artifact.importance, "source": "physical_scan", } ], ) artifact.layer = MemoryLayer.LONG_TERM artifact.memory_id = hashlib.sha256(artifact.content.encode()).hexdigest()[:16] self._artifact_registry[artifact.artifact_id] = artifact self._save_registry() return artifact.memory_id # ── Hive layer ──────────────────────────────────────────────── def mark_hive_validated(self, artifact_id: str, consensus: float): """Promote artifact to hive-validated status.""" if artifact_id in self._artifact_registry: art = self._artifact_registry[artifact_id] art.layer = MemoryLayer.HIVE art.hive_validated = True art.hive_consensus = consensus self._save_registry() # ── Queries ─────────────────────────────────────────────────── def list_artifacts( self, layer: Optional[MemoryLayer] = None ) -> List[MemoryArtifact]: arts = list(self._artifact_registry.values()) if layer: arts = [a for a in arts if a.layer == layer] return sorted(arts, key=lambda a: a.timestamp, reverse=True) def get_by_id(self, artifact_id: str) -> Optional[MemoryArtifact]: return self._artifact_registry.get(artifact_id) def stats(self) -> Dict[str, Any]: layers = {layer: 0 for layer in MemoryLayer} for a in self._artifact_registry.values(): layers[a.layer] += 1 return { "total_artifacts": len(self._artifact_registry), "by_layer": {k.name: v for k, v in layers.items()}, "hive_validated": sum( 1 for a in self._artifact_registry.values() if a.hive_validated ), "digital_memories": self.digital.count(), "artifact_dir": str(self.artifact_dir), } def export_all_cards(self) -> List[Path]: """Regenerate all physical cards from registry.""" paths = [] for art in self._artifact_registry.values(): path = self.artifact_dir / f"artifact_{art.artifact_id[:8]}.png" generate_qr_card(art, path, self.phi_logo_path) paths.append(path) return paths # ── CLI / quick test ────────────────────────────────────────────── if __name__ == "__main__": archive = MemoryArchive() print("Memory Archive initialized.") print(json.dumps(archive.stats(), indent=2))