import base64 import hashlib import io import json import os import time import uuid from collections import deque from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any import numpy as np import requests from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from PIL import Image APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) FRONTEND_INDEX = os.path.join(APP_ROOT, "frontend", "index.html") app = FastAPI(title="Live Code Camera Engine", version="0.2.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], # production: set exact origins allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) SYSTEM_PROMPT = """ You are Live Code Camera Engine. You generate code from a rolling sensor stream: - sampled camera frames - audio transcript text - scene-change metrics - signal-density metrics - user instruction You must output: 1. OBSERVATION: concise visible/audio evidence summary. 2. INFERENCE: what the evidence seems to imply, with uncertainty. 3. CODE: complete runnable code or patch. 4. RUN: exact run command. 5. RECEIPT: short audit record explaining which evidence hashes informed the output. Rules: - Do not claim to know hidden information. - Do not expose private chain-of-thought; provide a useful reasoning summary. - Do not execute generated code automatically. - Refuse malware, credential theft, covert surveillance, or unauthorized access. - If the scene includes faces or private environments, keep outputs privacy-preserving unless the user explicitly asks for local personal analysis. """ @dataclass class FrameEvidence: ts: float sha256: str phash: str entropy: float size_bytes: int width: int height: int delta: float jpeg_b64: str @dataclass class LiveSession: session_id: str = field(default_factory=lambda: str(uuid.uuid4())) frames: deque = field(default_factory=lambda: deque(maxlen=8)) transcripts: deque = field(default_factory=lambda: deque(maxlen=20)) created_at: float = field(default_factory=time.time) last_synthesis_at: float = 0.0 synth_count: int = 0 def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def sha256_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def parse_data_url(data_url: str) -> bytes: if "," in data_url and data_url.startswith("data:"): data_url = data_url.split(",", 1)[1] return base64.b64decode(data_url) def normalize_jpeg(image_bytes: bytes, max_side: int = 960) -> tuple[bytes, Image.Image]: im = Image.open(io.BytesIO(image_bytes)).convert("RGB") scale = min(1.0, max_side / max(im.width, im.height)) if scale < 1: im = im.resize((int(im.width * scale), int(im.height * scale))) out = io.BytesIO() im.save(out, format="JPEG", quality=82, optimize=True) return out.getvalue(), im def image_entropy(im: Image.Image) -> float: g = im.convert("L").resize((128, 128)) arr = np.asarray(g, dtype=np.uint8) hist = np.bincount(arr.flatten(), minlength=256).astype(np.float64) probs = hist / max(1, hist.sum()) probs = probs[probs > 0] return float(-(probs * np.log2(probs)).sum()) def average_hash(im: Image.Image, size: int = 8) -> str: g = im.convert("L").resize((size, size)) arr = np.asarray(g, dtype=np.float32) mean = arr.mean() bits = (arr > mean).astype(np.uint8).flatten() value = 0 for bit in bits: value = (value << 1) | int(bit) return f"{value:016x}" def phash_delta(a: str | None, b: str) -> float: if not a: return 1.0 x = int(a, 16) y = int(b, 16) return bin(x ^ y).count("1") / 64.0 def build_evidence_summary(session: LiveSession, user_instruction: str, mode: str) -> dict[str, Any]: frames = list(session.frames) transcript_text = " ".join(list(session.transcripts))[-4000:] if frames: avg_entropy = sum(f.entropy for f in frames) / len(frames) avg_delta = sum(f.delta for f in frames) / len(frames) last_frame = frames[-1] else: avg_entropy = 0 avg_delta = 0 last_frame = None return { "session_id": session.session_id, "mode": mode, "timestamp_utc": datetime.now(timezone.utc).isoformat(), "frame_count_buffered": len(frames), "frame_hashes": [f.sha256 for f in frames], "frame_phashes": [f.phash for f in frames], "avg_visual_entropy": round(avg_entropy, 4), "avg_frame_delta": round(avg_delta, 4), "transcript_hash": sha256_text(transcript_text), "instruction_hash": sha256_text(user_instruction), "transcript_excerpt": transcript_text[-1200:], "instruction": user_instruction, "last_frame": { "sha256": last_frame.sha256, "width": last_frame.width, "height": last_frame.height, "entropy": last_frame.entropy, "delta": last_frame.delta, } if last_frame else None, } def image_data_url_from_frame(frame: FrameEvidence) -> str: return "data:image/jpeg;base64," + frame.jpeg_b64 def call_ollama(prompt: str, frame: FrameEvidence | None) -> str: model = os.getenv("OLLAMA_MODEL", "llava") host = os.getenv("OLLAMA_HOST", "http://localhost:11434") payload = { "model": model, "prompt": prompt, "stream": False, } if frame: payload["images"] = [frame.jpeg_b64] r = requests.post(f"{host}/api/generate", json=payload, timeout=180) if r.status_code >= 400: raise RuntimeError(f"Ollama error: {r.text[:500]}") return r.json().get("response", "") def call_openai(prompt: str, frame: FrameEvidence | None) -> str: from openai import OpenAI model = os.getenv("OPENAI_MODEL") if not model: raise RuntimeError("OPENAI_MODEL is not set.") client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) content = [{"type": "input_text", "text": prompt}] if frame: content.append({"type": "input_image", "image_url": image_data_url_from_frame(frame)}) response = client.responses.create( model=model, input=[{"role": "user", "content": content}], ) return response.output_text def synthesize(session: LiveSession, user_instruction: str, mode: str) -> dict[str, Any]: evidence = build_evidence_summary(session, user_instruction, mode) last_frame = list(session.frames)[-1] if session.frames else None prompt = f""" {SYSTEM_PROMPT} MODE: {mode} USER INSTRUCTION: {user_instruction or "[none]"} ROLLING EVIDENCE SUMMARY: {json.dumps(evidence, indent=2)} Generate a live code update from the evidence. If the evidence is insufficient, generate the next instrumentation code needed to collect better evidence. """ provider = os.getenv("PROVIDER", "ollama").lower().strip() if provider == "ollama": output = call_ollama(prompt, last_frame) elif provider == "openai": output = call_openai(prompt, last_frame) else: raise RuntimeError(f"Unknown PROVIDER={provider!r}") session.synth_count += 1 session.last_synthesis_at = time.time() receipt = { "receipt_type": "LIVE_CODE_CAMERA_RECEIPT_V1", "session_id": session.session_id, "synthesis_index": session.synth_count, "provider": provider, "timestamp_utc": datetime.now(timezone.utc).isoformat(), "evidence_hash": sha256_text(json.dumps(evidence, sort_keys=True)), "prompt_hash": sha256_text(prompt), "frame_hashes": evidence["frame_hashes"], "transcript_hash": evidence["transcript_hash"], "avg_visual_entropy": evidence["avg_visual_entropy"], "avg_frame_delta": evidence["avg_frame_delta"], "mode": mode, } return {"type": "synthesis", "ok": True, "output": output, "receipt": receipt, "evidence": evidence} @app.get("/") def index(): return FileResponse(FRONTEND_INDEX) @app.get("/health") def health(): return { "ok": True, "app": "live-code-camera-engine", "provider": os.getenv("PROVIDER", "ollama"), "time": datetime.now(timezone.utc).isoformat(), } @app.websocket("/ws/live-code") async def live_code(ws: WebSocket): await ws.accept() session = LiveSession() await ws.send_json({"type": "session", "session_id": session.session_id}) try: while True: msg = await ws.receive_json() msg_type = msg.get("type") mode = msg.get("mode", "continuous_code") user_instruction = msg.get("instruction", "") if msg_type == "frame": raw = parse_data_url(msg["frame"]) jpeg, im = normalize_jpeg(raw) ph = average_hash(im) previous_ph = session.frames[-1].phash if session.frames else None evidence = FrameEvidence( ts=time.time(), sha256=sha256_bytes(jpeg), phash=ph, entropy=image_entropy(im), size_bytes=len(jpeg), width=im.width, height=im.height, delta=phash_delta(previous_ph, ph), jpeg_b64=base64.b64encode(jpeg).decode("ascii"), ) session.frames.append(evidence) await ws.send_json({ "type": "frame_ack", "sha256": evidence.sha256, "entropy": round(evidence.entropy, 4), "delta": round(evidence.delta, 4), "buffered": len(session.frames), }) auto = bool(msg.get("auto", True)) interval = float(msg.get("synthesis_interval", 8.0)) enough_time = time.time() - session.last_synthesis_at >= interval enough_frames = len(session.frames) >= 2 force = bool(msg.get("force", False)) if force or (auto and enough_time and enough_frames): try: result = synthesize(session, user_instruction, mode) await ws.send_json(result) except Exception as e: await ws.send_json({"type": "error", "message": str(e)}) elif msg_type == "transcript": text = (msg.get("text") or "").strip() if text: session.transcripts.append(text) await ws.send_json({"type": "transcript_ack", "chars": len(text)}) elif msg_type == "synthesize": try: result = synthesize(session, user_instruction, mode) await ws.send_json(result) except Exception as e: await ws.send_json({"type": "error", "message": str(e)}) elif msg_type == "ping": await ws.send_json({"type": "pong", "session_id": session.session_id}) else: await ws.send_json({"type": "error", "message": f"Unknown message type: {msg_type}"}) except WebSocketDisconnect: return