diff --git a/backend/api/main.py b/backend/api/main.py index 7a4ca4b..3312290 100644 --- a/backend/api/main.py +++ b/backend/api/main.py @@ -449,6 +449,17 @@ class WebhookCreateRequest(BaseModel): class SkillInstallRequest(BaseModel): name: str = Field(..., min_length=1, max_length=200) + +class MultimodalInput(BaseModel): + """Input for multimodal perception analysis (text + image + audio).""" + + text: str = Field(default="", max_length=50000) + image_base64: str | None = None # base64-encoded image + audio_base64: str | None = None # base64-encoded audio + media_type: str = Field(default="image/png", max_length=50) + agent_id: str | None = None + qalb_state: str = Field(default="", max_length=50) + # NOTE: Startup and shutdown logic is handled by the lifespan context manager above. @@ -1082,6 +1093,46 @@ async def list_memories(memory_type: str | None = None, limit: int = 30): return {"results": results, "total": len(results)} +# === PERCEPTION (Sam' + Basar - سَمْع + بَصَر) === + + +@app.post("/api/perception/analyze") +async def analyze_multimodal( + req: MultimodalInput, + user: TokenPayload | None = Depends(get_current_user), +): + """Analyze multimodal input through the QCA perception pipeline. + + Accepts text, base64-encoded images, and base64-encoded audio. + Follows Quranic ordering: Sam' (hearing) before Basar (sight). + """ + import base64 as b64 + + agent = None + if req.agent_id and req.agent_id in active_agents: + agent = active_agents[req.agent_id] + elif active_agents: + agent = next(iter(active_agents.values())) + + if not agent: + raise HTTPException(503, "No agents available for perception analysis") + + image_bytes = b64.b64decode(req.image_base64) if req.image_base64 else None + audio_bytes = b64.b64decode(req.audio_base64) if req.audio_base64 else None + + if not req.text and not image_bytes and not audio_bytes: + raise HTTPException(400, "At least one of text, image_base64, or audio_base64 is required") + + result = await agent.qca.process_input_multimodal( + text=req.text, + image_bytes=image_bytes, + audio_bytes=audio_bytes, + media_type=req.media_type, + qalb_state=req.qalb_state, + ) + return {"result": result} + + # === KNOWLEDGE INGESTION (Ilm - عِلْم) === @@ -2580,6 +2631,51 @@ async def task_stream(chunk): }, ) + elif msg_type == "multimodal": + import base64 as b64 + + session_id = data.get("session_id", client_id) + text = data.get("content", "") + image_b64 = data.get("image_base64") + audio_b64 = data.get("audio_base64") + media_type = data.get("media_type", "image/png") + qalb_state = data.get("qalb_state", "") + + image_bytes = b64.b64decode(image_b64) if image_b64 else None + audio_bytes = b64.b64decode(audio_b64) if audio_b64 else None + + agent = next(iter(active_agents.values()), None) if active_agents else None + if agent: + try: + result = await agent.qca.process_input_multimodal( + text=text, + image_bytes=image_bytes, + audio_bytes=audio_bytes, + media_type=media_type, + qalb_state=qalb_state, + ) + await manager.send( + client_id, + { + "type": "perception_result", + "session_id": session_id, + "result": result, + }, + ) + except Exception as e: + await manager.send( + client_id, + { + "type": "error", + "message": f"Multimodal processing failed: {e}", + }, + ) + else: + await manager.send( + client_id, + {"type": "error", "message": "No agents available"}, + ) + elif msg_type == "command": cmd = data.get("command", "").strip() if cmd == "/status" or cmd == "status": diff --git a/backend/memory/knowledge_graph.py b/backend/memory/knowledge_graph.py index 709c3e4..fb5ee86 100644 --- a/backend/memory/knowledge_graph.py +++ b/backend/memory/knowledge_graph.py @@ -178,6 +178,50 @@ async def query_entity(self, name: str) -> dict: "incoming": incoming, } + async def search_entities( + self, query: str, limit: int = 10, entity_type: str | None = None + ) -> list[dict]: + """Search entities by name (LIKE query) with optional type filtering. + + Used by MemoryPyramid layer 4 for graph-based recall. + """ + conn = sqlite3.connect(self.db_path) + c = conn.cursor() + + results = [] + seen_ids: set[str] = set() + words = query.lower().split() + + for word in words[:5]: + if len(word) < 2: + continue + if entity_type: + c.execute( + "SELECT id, name, type, properties FROM kg_entities " + "WHERE name LIKE ? AND type = ? LIMIT ?", + (f"%{word}%", entity_type, limit), + ) + else: + c.execute( + "SELECT id, name, type, properties FROM kg_entities " + "WHERE name LIKE ? LIMIT ?", + (f"%{word}%", limit), + ) + for row in c.fetchall(): + if row[0] not in seen_ids: + seen_ids.add(row[0]) + results.append( + { + "id": row[0], + "name": row[1], + "type": row[2], + "properties": json.loads(row[3]) if row[3] else {}, + } + ) + + conn.close() + return results[:limit] + async def get_stats(self) -> dict: """Get knowledge graph statistics""" conn = sqlite3.connect(self.db_path) diff --git a/backend/memory/lawh_mahfuz.py b/backend/memory/lawh_mahfuz.py index 9ef6067..11f4b8a 100644 --- a/backend/memory/lawh_mahfuz.py +++ b/backend/memory/lawh_mahfuz.py @@ -37,6 +37,7 @@ class LawhEntry: length: int # len(content) in bytes certainty: float = 1.0 category: str = "general" + quaternary_checksum: str = "" # DNA-inspired quaternary checksum (ACGT) def to_dict(self) -> dict: return { @@ -46,6 +47,7 @@ def to_dict(self) -> dict: "certainty": self.certainty, "category": self.category, "stored_at": self.stored_at, + "quaternary_checksum": self.quaternary_checksum, } @@ -108,6 +110,11 @@ def _init_db(self): quarantined_at REAL ) """) + # Migration: add quaternary_checksum column if not present + try: + c.execute("ALTER TABLE lawh_entries ADD COLUMN quaternary_checksum TEXT DEFAULT ''") + except sqlite3.OperationalError: + pass # Column already exists conn.commit() finally: conn.close() @@ -119,11 +126,13 @@ def _load_into_cache(self): c = conn.cursor() c.execute("SELECT * FROM lawh_entries") for row in c.fetchall(): - key, content, source, stored_at, sha256, crc32, length, certainty, category = row + key, content, source, stored_at, sha256, crc32, length, certainty, category = row[:9] + quat_checksum = row[9] if len(row) > 9 else "" entry = LawhEntry( key=key, content=content, source=source, stored_at=stored_at, sha256=sha256, crc32=crc32, length=length, certainty=certainty, category=category, + quaternary_checksum=quat_checksum or "", ) self._cache[key] = entry logger.info("[LAWH] Loaded %d entries from preserved tablet", len(self._cache)) @@ -152,10 +161,19 @@ def store_immutable( length = len(content.encode("utf-8")) now = time.time() + # Compute quaternary checksum (DNA-inspired integrity layer) + quat_checksum = "" + try: + from memory.quaternary import quaternary_checksum + quat_checksum = quaternary_checksum(content) + except ImportError: + pass + entry = LawhEntry( key=key, content=content, source=source, stored_at=now, sha256=sha, crc32=crc, length=length, certainty=certainty, category=category, + quaternary_checksum=quat_checksum, ) conn = self._get_conn() @@ -163,16 +181,16 @@ def store_immutable( c = conn.cursor() c.execute( """INSERT OR IGNORE INTO lawh_entries - (key, content, source, stored_at, sha256, crc32, length, certainty, category) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (key, content, source, now, sha, crc, length, certainty, category), + (key, content, source, stored_at, sha256, crc32, length, certainty, category, quaternary_checksum) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (key, content, source, now, sha, crc, length, certainty, category, quat_checksum), ) conn.commit() finally: conn.close() self._cache[key] = entry - logger.debug("[LAWH] Stored '%s' (len=%d sha=%s...)", key, length, sha[:8]) + logger.debug("[LAWH] Stored '%s' (len=%d sha=%s... quat=%s...)", key, length, sha[:8], quat_checksum[:8] if quat_checksum else "n/a") return key def verify_integrity(self, key: str) -> bool: @@ -203,6 +221,20 @@ def verify_integrity(self, key: str) -> bool: self._quarantine_entry(key, f"Length mismatch: {actual_len}≠{entry.length}") return False + # 4th check: Quaternary checksum (DNA-inspired, when present) + if entry.quaternary_checksum: + try: + from memory.quaternary import hamming_distance, quaternary_checksum + actual_quat = quaternary_checksum(entry.content) + dist = hamming_distance(actual_quat, entry.quaternary_checksum) + if dist > 0: + self._quarantine_entry( + key, f"Quaternary checksum drift: hamming_distance={dist}" + ) + return False + except ImportError: + pass # Quaternary module not available; skip this check + return True def get(self, key: str) -> LawhEntry | None: diff --git a/backend/memory/living_memory.py b/backend/memory/living_memory.py index 5a3a0ff..1e0577b 100644 --- a/backend/memory/living_memory.py +++ b/backend/memory/living_memory.py @@ -167,7 +167,7 @@ class LivingMemorySystem: context-dependent recall, and Dhikr maintenance daemon. """ - def __init__(self, masalik=None, dhikr_db=None, lawh=None): + def __init__(self, masalik=None, dhikr_db=None, lawh=None, vector_store=None): # Memory stores by level self.sadr: list[MemoryTrace] = [] # working memory (capacity-limited) self.traces: dict[str, MemoryTrace] = {} # all traces by ID @@ -177,6 +177,7 @@ def __init__(self, masalik=None, dhikr_db=None, lawh=None): self._masalik = masalik # MasalikNetwork for spread activation self._dhikr_db = dhikr_db # DhikrMemorySystem for persistence self._lawh = lawh # LawhMahfuz for immutable storage + self._vector_store = vector_store # VectorStore for semantic similarity self._daemon_cycle = 0 self._content_hashes: dict[str, str] = {} # hash → trace_id for fast lookup @@ -620,10 +621,31 @@ def _store_trace(self, trace: MemoryTrace) -> None: except Exception: pass + # Persist to VectorStore for semantic similarity search + if self._vector_store and getattr(self._vector_store, "_available", False): + try: + import asyncio + import concurrent.futures + + metadata = {"level": trace.level.value, "importance": trace.importance} + coro = self._vector_store.store( + trace.content, memory_id=trace.trace_id, metadata=metadata + ) + try: + asyncio.get_running_loop() + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + executor.submit(asyncio.run, coro).result(timeout=5) + except RuntimeError: + asyncio.run(coro) + except Exception: + pass + def _find_best_match(self, content: str) -> tuple[MemoryTrace | None, float]: - """Find the most similar existing trace.""" + """Find the most similar existing trace (text + optional vector similarity).""" best = None best_sim = 0.0 + + # Primary: Jaccard text similarity (always available) for trace in self.traces.values(): if trace.trace_id in self.archive: continue @@ -631,8 +653,49 @@ def _find_best_match(self, content: str) -> tuple[MemoryTrace | None, float]: if sim > best_sim: best_sim = sim best = trace + + # Secondary: Vector similarity via ChromaDB (if available) + if self._vector_store and getattr(self._vector_store, "_available", False): + try: + # VectorStore.search() is async but underlying ChromaDB is sync. + # Use sync wrapper to avoid event loop issues. + vector_results = self._vector_search_sync(content, limit=5) + for vr in vector_results: + distance = vr.get("distance", 1.0) + # ChromaDB L2 distance → 0-1 similarity + vector_sim = max(0.0, 1.0 - (distance / 2.0)) + vec_id = vr.get("id", "") + if vec_id in self.traces and vec_id not in self.archive: + trace = self.traces[vec_id] + hybrid_sim = max( + self._text_similarity(content, trace.content), + vector_sim, + ) + if hybrid_sim > best_sim: + best_sim = hybrid_sim + best = trace + except Exception: + pass # Graceful degradation to text-only + return best, best_sim + def _vector_search_sync(self, query: str, limit: int = 5) -> list[dict]: + """Synchronous wrapper for VectorStore.search().""" + import asyncio + import concurrent.futures + + try: + asyncio.get_running_loop() + # Already in async context — run in thread to avoid nested loop + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + asyncio.run, self._vector_store.search(query, limit=limit) + ) + return future.result(timeout=5) + except RuntimeError: + # No running event loop — safe to run directly + return asyncio.run(self._vector_store.search(query, limit=limit)) + def _extract_novel_parts(self, new_content: str, existing: MemoryTrace) -> str: """Extract what's genuinely new in content vs existing trace.""" new_words = set(new_content.lower().split()) diff --git a/backend/memory/quaternary.py b/backend/memory/quaternary.py new file mode 100644 index 0000000..3cf266a --- /dev/null +++ b/backend/memory/quaternary.py @@ -0,0 +1,235 @@ +""" +Quaternary Encoding (تشفير رباعي) — DNA-Inspired Data Integrity +================================================================ + +"And We created you in pairs (azwaj)" — Quran 78:8 +"Read in the name of your Lord who created — created man from a clinging substance" — Quran 96:1-2 + +Binary data encoded using a quaternary alphabet (A, C, G, T), +mirroring DNA's information storage: +- 2 bits → 1 quaternary symbol +- 3 symbols → 1 codon (6 bits → 64 possible codons) +- Codons map to semantic categories + +Error detection inspired by biological error correction: +- Per-codon XOR parity symbol +- Hamming distance for corruption detection +- Quaternary checksums for integrity verification +""" + +import hashlib + +# ─── Quaternary Alphabet ────────────────────────────────────────────────────── +# Maps 2-bit pairs to nucleotide symbols (like DNA base pairs) +QUAT_MAP = {0b00: "A", 0b01: "C", 0b10: "G", 0b11: "T"} +QUAT_REVERSE = {"A": 0b00, "C": 0b01, "G": 0b10, "T": 0b11} +VALID_SYMBOLS = frozenset("ACGT") + + +# ─── Encoding / Decoding ───────────────────────────────────────────────────── + + +def bytes_to_quaternary(data: bytes) -> str: + """Convert binary data to quaternary string (A/C/G/T). + + Each byte produces 4 quaternary symbols (2 bits per symbol). + + >>> bytes_to_quaternary(b'\\x00') + 'AAAA' + >>> bytes_to_quaternary(b'\\xff') + 'TTTT' + """ + result = [] + for byte in data: + result.append(QUAT_MAP[(byte >> 6) & 0b11]) + result.append(QUAT_MAP[(byte >> 4) & 0b11]) + result.append(QUAT_MAP[(byte >> 2) & 0b11]) + result.append(QUAT_MAP[byte & 0b11]) + return "".join(result) + + +def quaternary_to_bytes(quat: str) -> bytes: + """Convert quaternary string back to binary data. + + >>> quaternary_to_bytes('AAAA') + b'\\x00' + >>> quaternary_to_bytes('TTTT') + b'\\xff' + """ + # Validate input + if not all(c in VALID_SYMBOLS for c in quat): + raise ValueError("Invalid quaternary string: contains non-ACGT characters") + + # Pad to multiple of 4 + padded = quat + while len(padded) % 4 != 0: + padded += "A" + + result = bytearray() + for i in range(0, len(padded), 4): + byte = ( + (QUAT_REVERSE[padded[i]] << 6) + | (QUAT_REVERSE[padded[i + 1]] << 4) + | (QUAT_REVERSE[padded[i + 2]] << 2) + | QUAT_REVERSE[padded[i + 3]] + ) + result.append(byte) + return bytes(result) + + +# ─── Codon Chunking ────────────────────────────────────────────────────────── +# 3 quaternary symbols = 1 codon (64 possible triplets → semantic categories) + +CODON_CATEGORIES = [ + "data", "reference", "separator", "checksum", "metadata", "padding", +] + + +def _build_codon_table() -> dict[str, str]: + """Build complete 64-codon to category mapping.""" + alphabet = "ACGT" + table: dict[str, str] = {} + idx = 0 + for a in alphabet: + for b in alphabet: + for c in alphabet: + codon = a + b + c + if codon == "AAA": + table[codon] = "start" + elif codon == "TTT": + table[codon] = "stop" + else: + table[codon] = CODON_CATEGORIES[idx % len(CODON_CATEGORIES)] + idx += 1 + return table + + +CODON_TABLE = _build_codon_table() + + +def to_codons(quat_string: str) -> list[str]: + """Split quaternary string into triplet codons. + + >>> to_codons('ACGTAC') + ['ACG', 'TAC'] + """ + # Pad to multiple of 3 + padded = quat_string + while len(padded) % 3 != 0: + padded += "A" + return [padded[i : i + 3] for i in range(0, len(padded), 3)] + + +def classify_codons(codons: list[str]) -> list[tuple[str, str]]: + """Classify each codon into its semantic category. + + >>> classify_codons(['AAA', 'TTT']) + [('AAA', 'start'), ('TTT', 'stop')] + """ + return [(codon, CODON_TABLE.get(codon, "data")) for codon in codons] + + +# ─── Error Detection (Parity) ──────────────────────────────────────────────── + + +def compute_parity_symbol(codon: str) -> str: + """Compute a parity symbol for a codon (XOR of three symbols). + + >>> compute_parity_symbol('ACG') + 'C' + """ + if len(codon) != 3 or not all(c in VALID_SYMBOLS for c in codon): + raise ValueError(f"Invalid codon: {codon!r}") + values = [QUAT_REVERSE[c] for c in codon] + parity = values[0] ^ values[1] ^ values[2] + return QUAT_MAP[parity & 0b11] + + +def encode_with_parity(quat_string: str) -> str: + """Encode quaternary string with per-codon parity symbols. + + Every 3 data symbols get 1 parity symbol appended (3+1=4 per block). + """ + codons = to_codons(quat_string) + result = [] + for codon in codons: + parity = compute_parity_symbol(codon) + result.append(codon + parity) + return "".join(result) + + +def verify_and_correct(encoded: str) -> tuple[str, bool, int]: + """Verify parity and detect errors. + + Returns (data_without_parity, is_valid, errors_detected). + """ + blocks = [encoded[i : i + 4] for i in range(0, len(encoded), 4)] + corrected_data = [] + errors = 0 + + for block in blocks: + if len(block) < 4: + corrected_data.append(block[:3] if len(block) >= 3 else block) + continue + + codon = block[:3] + stored_parity = block[3] + expected_parity = compute_parity_symbol(codon) + + if stored_parity == expected_parity: + corrected_data.append(codon) + else: + errors += 1 + # Error detected — pass through data (correction requires + # additional redundancy beyond single parity) + corrected_data.append(codon) + + return "".join(corrected_data), errors == 0, errors + + +# ─── Checksums ──────────────────────────────────────────────────────────────── + + +def quaternary_checksum(content: str) -> str: + """Generate a quaternary checksum for text content. + + Takes SHA-256 of content, converts first 8 bytes to 32 quaternary symbols. + + >>> len(quaternary_checksum("hello")) + 32 + >>> all(c in 'ACGT' for c in quaternary_checksum("hello")) + True + """ + sha = hashlib.sha256(content.encode("utf-8")).digest() + return bytes_to_quaternary(sha[:8]) # 8 bytes → 32 quaternary symbols + + +def hamming_distance(a: str, b: str) -> int: + """Compute Hamming distance between two quaternary strings. + + Counts the number of symbol positions where the two strings differ. + + >>> hamming_distance('ACGT', 'ACGT') + 0 + >>> hamming_distance('ACGT', 'TCGA') + 4 + """ + return sum(1 for x, y in zip(a, b, strict=False) if x != y) + + +def verify_checksum(content: str, stored_checksum: str, tolerance: int = 0) -> bool: + """Verify content against a stored quaternary checksum. + + Args: + content: The text content to verify. + stored_checksum: The quaternary checksum to verify against. + tolerance: Maximum Hamming distance allowed (0 = exact match). + + Returns: + True if the content matches within tolerance. + """ + if not stored_checksum: + return True # No checksum to verify against + actual = quaternary_checksum(content) + dist = hamming_distance(actual, stored_checksum) + return dist <= tolerance diff --git a/backend/perception/__init__.py b/backend/perception/__init__.py index 668c43e..e43604f 100644 --- a/backend/perception/__init__.py +++ b/backend/perception/__init__.py @@ -7,5 +7,7 @@ Multimodal perception: voice, vision, and document understanding. """ +from .basirah import BasirahEngine as BasirahEngine +from .nutq import NutqEngine as NutqEngine from .vision import VisionProcessor as VisionProcessor from .voice import VoiceProcessor as VoiceProcessor diff --git a/backend/perception/basirah.py b/backend/perception/basirah.py index b409f46..9543c6f 100644 --- a/backend/perception/basirah.py +++ b/backend/perception/basirah.py @@ -13,7 +13,9 @@ - Qalb-aware interpretation (considers emotional context) """ +import json as _json import logging +import re import time from dataclasses import dataclass, field @@ -62,62 +64,114 @@ def __init__(self): self._vision = VisionProcessor() async def analyze( - self, image_bytes: bytes, context: str = "", media_type: str = "image/png" + self, + image_bytes: bytes, + context: str = "", + media_type: str = "image/png", + qalb_state: str = "", ) -> BasirahInsight: - """Analyze an image with contextual insight.""" + """Analyze an image with contextual insight, modulated by emotional state.""" start = time.time() - prompt = self._build_prompt(context) + prompt = self._build_prompt(context, qalb_state) raw_result = await self._vision.analyze_image(image_bytes, prompt, media_type) - # Parse the result into structured insight insight = self._parse_result(raw_result) insight.processing_time_ms = (time.time() - start) * 1000 return insight - async def analyze_url(self, url: str, context: str = "") -> BasirahInsight: + async def analyze_url( + self, url: str, context: str = "", qalb_state: str = "" + ) -> BasirahInsight: """Analyze an image from URL with contextual insight.""" start = time.time() - prompt = self._build_prompt(context) + prompt = self._build_prompt(context, qalb_state) raw_result = await self._vision.analyze_image_url(url, prompt) insight = self._parse_result(raw_result) insight.processing_time_ms = (time.time() - start) * 1000 return insight - def _build_prompt(self, context: str) -> str: + def _build_prompt(self, context: str, qalb_state: str = "") -> str: base = ( - "Analyze this image and provide:\n" - "1. A clear description of what you see\n" - "2. Category: text, diagram, screenshot, photo, or document\n" - "3. Any text visible in the image\n" - "4. Key elements or objects identified\n" + "Analyze this image and respond in the following JSON format:\n" + '{"description": "what you see in the image", ' + '"category": "text|diagram|screenshot|photo|document", ' + '"extracted_text": "any text visible in the image", ' + '"key_elements": ["element1", "element2"], ' + '"confidence": 0.0}\n' + "Set confidence between 0.0 and 1.0 based on your certainty.\n" ) if context: base += f"\nContext: {context}" + + # Qalb-aware perception: adjust focus based on emotional context + if qalb_state and qalb_state != "neutral": + focus_map = { + "frustrated": "Pay special attention to error messages, warnings, or problematic elements.", + "confused": "Focus on clarity — highlight text, labels, and structural elements.", + "anxious": "Note any reassuring or concerning elements. Be thorough but gentle.", + "determined": "Focus on actionable information and key data points.", + "positive": "Highlight achievements, successes, and positive indicators.", + } + focus = focus_map.get(qalb_state, "") + if focus: + base += f"\n{focus}" + return base def _parse_result(self, raw: str) -> BasirahInsight: """Parse raw vision output into structured insight.""" - # Determine category from content - raw_lower = raw.lower() - if any(w in raw_lower for w in ["screenshot", "interface", "ui", "window", "browser"]): - category = "screenshot" - elif any(w in raw_lower for w in ["diagram", "chart", "graph", "flow"]): - category = "diagram" - elif any(w in raw_lower for w in ["document", "page", "pdf", "form"]): - category = "document" - elif any(w in raw_lower for w in ["error" in raw_lower and "text"]): - category = "text" - else: - category = "photo" - + if not raw: + return BasirahInsight( + description="No result from vision analysis", + category="photo", + confidence=0.0, + ) + + # Try to extract JSON from LLM response + try: + json_match = re.search(r"\{[^{}]*\}", raw, re.DOTALL) + if json_match: + parsed = _json.loads(json_match.group()) + return BasirahInsight( + description=str(parsed.get("description", raw[:500]))[:500], + category=self._validate_category(parsed.get("category", "")), + confidence=min(1.0, max(0.0, float(parsed.get("confidence", 0.7)))), + extracted_text=str(parsed.get("extracted_text", ""))[:2000], + key_elements=[str(e) for e in parsed.get("key_elements", [])][:20], + ) + except (ValueError, KeyError, TypeError): + pass + + # Fallback: heuristic parsing return BasirahInsight( description=raw[:500], - category=category, - confidence=0.8, - extracted_text="", # Would be populated by more advanced parsing + category=self._guess_category(raw), + confidence=0.6, # Lower confidence for unparsed results + extracted_text="", key_elements=[], ) + + @staticmethod + def _validate_category(category: str) -> str: + """Validate and normalize the category string.""" + valid = {"text", "diagram", "screenshot", "photo", "document"} + cat = category.lower().strip() + return cat if cat in valid else "photo" + + @staticmethod + def _guess_category(raw: str) -> str: + """Heuristic category detection from raw text.""" + raw_lower = raw.lower() + if any(w in raw_lower for w in ["screenshot", "interface", "ui", "window", "browser"]): + return "screenshot" + if any(w in raw_lower for w in ["diagram", "chart", "graph", "flow"]): + return "diagram" + if any(w in raw_lower for w in ["document", "page", "pdf", "form"]): + return "document" + if "error" in raw_lower and "text" in raw_lower: + return "text" + return "photo" diff --git a/backend/perception/nutq.py b/backend/perception/nutq.py index 76604db..e211a46 100644 --- a/backend/perception/nutq.py +++ b/backend/perception/nutq.py @@ -81,26 +81,43 @@ def __init__(self): self._voice = VoiceProcessor() async def speak( - self, text: str, tone: str = "standard", voice: str = "default" + self, + text: str, + tone: str = "standard", + voice: str = "default", + qalb_state: str = "", ) -> NutqUtterance: - """Generate speech with tone calibration.""" + """Generate speech with tone calibration, optionally driven by Qalb state.""" start = time.time() - # Adjust text for tone (light modifications) - adjusted_text = self._adjust_for_tone(text, tone) - + # Qalb-driven tone override (when tone is default "standard") + effective_tone = tone + if qalb_state and tone == "standard": + qalb_tone_map = { + "frustrated": "patient", + "anxious": "warm", + "confused": "patient", + "fatigued": "warm", + "determined": "focused", + "positive": "warm", + } + effective_tone = qalb_tone_map.get(qalb_state, tone) + + adjusted_text = self._adjust_for_tone(text, effective_tone) + language = self._detect_language(adjusted_text) audio = await self._voice.text_to_speech(adjusted_text, voice) elapsed = (time.time() - start) * 1000 return NutqUtterance( text=adjusted_text, audio_bytes=audio, - tone=tone, + tone=effective_tone, + language=language, duration_ms=elapsed, ) async def listen(self, audio_bytes: bytes) -> NutqTranscription: - """Transcribe speech with intent detection.""" + """Transcribe speech with intent and language detection.""" start = time.time() text = await self._voice.speech_to_text(audio_bytes) @@ -112,41 +129,117 @@ async def listen(self, audio_bytes: bytes) -> NutqTranscription: ) intent = self._detect_intent(text) + language = self._detect_language(text) elapsed = (time.time() - start) * 1000 return NutqTranscription( text=text, confidence=0.9, + detected_language=language, detected_intent=intent, processing_time_ms=elapsed, ) - def _adjust_for_tone(self, text: str, tone: str) -> str: + @staticmethod + def _adjust_for_tone(text: str, tone: str) -> str: """Light text adjustments based on desired tone.""" - # In a full implementation, this would modify pacing markers, - # emphasis, and SSML tags. For now, return as-is. + if not text: + return text + + if tone == "warm": + # Softer pacing: add pause markers between sentences + text = text.replace(". ", "... ").replace("! ", "... ") + elif tone == "patient": + # Slow down: extra spacing between sentences + text = text.replace(". ", ". ").replace("? ", "? ") + elif tone == "focused": + # Strip filler words for conciseness + fillers = ["well, ", "so, ", "you know, ", "basically, ", "actually, ", "like, "] + for filler in fillers: + text = text.replace(filler, "") + text = text.replace(filler.capitalize(), "") + return text - def _detect_intent(self, text: str) -> str: + @staticmethod + def _detect_intent(text: str) -> str: """Detect the intent of transcribed speech.""" text_stripped = text.strip() + if not text_stripped: + return "statement" + + text_lower = text_stripped.lower() + + # Question detection: question mark or question words at start if text_stripped.endswith("?"): return "question" - if any( - text_stripped.lower().startswith(w) - for w in [ - "do", - "run", - "create", - "delete", - "open", - "close", - "find", - "search", - "show", - "tell", - "help", - ] - ): + question_starts = [ + "what ", "where ", "when ", "who ", "whom ", "which ", "why ", "how ", + "is it ", "are there ", "can you ", "could you ", "would you ", + "do you ", "does it ", "will it ", "shall ", + ] + if any(text_lower.startswith(q) for q in question_starts): + return "question" + + # Greeting detection + greetings = [ + "hello", "hi ", "hey ", "good morning", "good afternoon", + "good evening", "salam", "assalamu", + ] + if any(text_lower.startswith(g) for g in greetings): + return "greeting" + + # Farewell detection + farewells = ["goodbye", "bye", "see you", "take care", "good night", "ma'a salama"] + if any(text_lower.startswith(f) for f in farewells): + return "farewell" + + # Confirmation/negation + if text_lower in ("yes", "yeah", "yep", "sure", "ok", "okay", "right", "correct"): + return "confirmation" + if text_lower in ("no", "nope", "nah", "wrong", "incorrect"): + return "negation" + + # Command detection: imperative verbs + command_verbs = [ + "do ", "run ", "create ", "delete ", "open ", "close ", "find ", "search ", + "show ", "tell ", "help ", "make ", "build ", "write ", "read ", "send ", + "stop ", "start ", "restart ", "install ", "update ", "fix ", "check ", + "list ", "get ", "set ", "add ", "remove ", "move ", "copy ", "save ", + "load ", "deploy ", "test ", "analyze ", "explain ", "summarize ", + ] + if any(text_lower.startswith(w) for w in command_verbs): return "command" + + # Request detection (polite commands) + request_patterns = ["please ", "could you ", "can you ", "would you ", "i need ", "i want "] + if any(text_lower.startswith(p) for p in request_patterns): + return "request" + return "statement" + + @staticmethod + def _detect_language(text: str) -> str: + """Basic language detection based on Unicode character ranges.""" + if not text: + return "en" + + # Count characters in Arabic Unicode range + arabic_count = sum(1 for c in text if "\u0600" <= c <= "\u06FF") + # Extended Arabic (includes Urdu-specific) + extended_arabic = sum(1 for c in text if "\uFB50" <= c <= "\uFDFF" or "\uFE70" <= c <= "\uFEFF") + total_alpha = sum(1 for c in text if c.isalpha()) + + if total_alpha == 0: + return "en" + + arabic_ratio = (arabic_count + extended_arabic) / total_alpha + + if arabic_ratio > 0.3: + # Distinguish Arabic from Urdu by checking for Urdu-specific characters + urdu_specific = sum(1 for c in text if c in "\u0679\u067E\u0686\u0688\u0691\u0698\u06BA\u06BE\u06C1\u06CC\u06D2") + if urdu_specific > 0: + return "ur" + return "ar" + + return "en" diff --git a/backend/qca/engine.py b/backend/qca/engine.py index 43e0fae..b141b44 100644 --- a/backend/qca/engine.py +++ b/backend/qca/engine.py @@ -137,6 +137,73 @@ def process(self, text: str) -> dict: fuad = self.integrate_fuad(sam, basar) return {"sam": sam, "basar": basar, "fuad": fuad} + async def process_multimodal( + self, + text: str = "", + image_bytes: bytes | None = None, + audio_bytes: bytes | None = None, + media_type: str = "image/png", + context: str = "", + qalb_state: str = "", + ) -> dict: + """ + Full multimodal processing: Sam' (hearing) first, then Basar (sight), + following the Quranic ordering of 16:78 and 17:36. + + "And Allah brought you out from the wombs of your mothers while you knew + nothing, and He gave you hearing (sam'), sight (basar), and hearts (af'ida) + that perhaps you would be grateful." — Quran 16:78 + """ + from perception.basirah import BasirahEngine + from perception.nutq import NutqEngine + + results: dict = {"sam": {}, "basar": {}, "fuad": {}, "nutq": None, "basirah": None} + + # ── Step 1: Sam' — auditory input first (hearing precedes sight) ── + nutq_text = "" + if audio_bytes: + try: + nutq = NutqEngine() + transcription = await nutq.listen(audio_bytes) + results["nutq"] = transcription.to_dict() + nutq_text = transcription.text + except Exception as e: + logger.warning("[QCA] Nutq (audio) processing failed: %s", e) + + # Combine text sources: explicit text + transcribed speech + combined_text = " ".join(filter(None, [text, nutq_text])) + + # ── Step 2: Basar — visual input second ── + if image_bytes: + try: + basirah = BasirahEngine() + insight = await basirah.analyze( + image_bytes, context=context, media_type=media_type, + qalb_state=qalb_state, + ) + results["basirah"] = insight.to_dict() + # Append extracted text from vision to combined text + if insight.extracted_text: + combined_text += " " + insight.extracted_text + except Exception as e: + logger.warning("[QCA] Basirah (vision) processing failed: %s", e) + + # ── Step 3: Fu'ad — standard text integration on combined input ── + if combined_text.strip(): + sam = self.process_sam(combined_text) + basar = self.process_basar(combined_text) + fuad = self.integrate_fuad(sam, basar) + results["sam"] = sam + results["basar"] = basar + results["fuad"] = fuad + else: + results["fuad"] = { + "zahir": "", "batin": "", "key_terms": [], + "vocabulary_richness": 0, "sequential_pairs": [], "total_tokens": 0, + } + + return results + # ───────────────────────────────────────────────────────────────────────────── # LAYER 4: ISM — Root-Space Semantic Representation @@ -821,6 +888,71 @@ def process_input(self, text: str) -> dict: "batin": perception["fuad"]["batin"], } + async def process_input_multimodal( + self, + text: str = "", + image_bytes: bytes | None = None, + audio_bytes: bytes | None = None, + media_type: str = "image/png", + context: str = "", + qalb_state: str = "", + ) -> dict: + """ + Process multimodal input through perception layers + ISM. + + Follows Quranic ordering: Sam' (hearing) first, then Basar (sight). + Falls back to text-only process_input() if no multimodal data present. + """ + if not image_bytes and not audio_bytes: + return self.process_input(text) + + perception = await self.dual_input.process_multimodal( + text=text, + image_bytes=image_bytes, + audio_bytes=audio_bytes, + media_type=media_type, + context=context, + qalb_state=qalb_state, + ) + + # Combine all text sources for root analysis + combined_text = text + nutq_result = perception.get("nutq") + if nutq_result and nutq_result.get("text"): + combined_text += " " + nutq_result["text"] + basirah_result = perception.get("basirah") + if basirah_result and basirah_result.get("extracted_text"): + combined_text += " " + basirah_result["extracted_text"] + + roots = self.ism.find_roots_in_text(combined_text) if combined_text.strip() else {} + + # Store in Tier 3 working memory + self.lawh.store( + "CURRENT_INPUT", + combined_text[:500], + certainty=1.0, + source="multimodal_input", + tier=3, + ) + key_terms = perception.get("fuad", {}).get("key_terms", []) + self.lawh.store( + "CURRENT_TERMS", + str(key_terms), + certainty=1.0, + source="fuad_analysis", + tier=3, + ) + + return { + "perception": perception, + "roots_identified": roots, + "key_terms": key_terms, + "zahir": perception.get("fuad", {}).get("zahir", ""), + "batin": perception.get("fuad", {}).get("batin", ""), + "has_vision": image_bytes is not None, + "has_audio": audio_bytes is not None, + } + def reason(self, question: str, context_text: str = None) -> dict: """ Full QCA reasoning pipeline for answering a question.