Spaces:

josephrw
/

live-code-camera-engine

Paused

live-code-camera-engine / backend /main.py

Upload folder using huggingface_hub

3132104 verified 5 months ago

11.4 kB

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

	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