Spaces:
Building
Building
File size: 7,631 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | import logging
async def xr_anchor_tool(action: str, anchor_id: str = "", *args, **kwargs):
"""
Control or query XR spatial anchors.
Action: 'save', 'get', 'clear'
"""
from backend.ws.agent_ws import ws_manager
await ws_manager.broadcast({
"event": "xr:anchor",
"payload": {"action": action, "anchor_id": anchor_id}
})
return f"Broadcasted XR anchor command: {action} on {anchor_id}"
async def gesture_context_tool(gesture: str, *args, **kwargs):
"""
Fake/inject an XR gesture for testing or context (e.g. 'pinch', 'swipe').
"""
from backend.ws.agent_ws import ws_manager
await ws_manager.broadcast({
"event": f"gesture-{gesture}",
"payload": {"source": "jarvis"}
})
return f"Injected simulated gesture: {gesture}"
async def ar_control_tool(action: str, payload: str = "", *args, **kwargs):
"""
Control AR states globally (e.g. toggle AR, spawn model, enter vr, clear scene).
"""
from backend.ws.agent_ws import ws_manager
await ws_manager.broadcast({
"event": "ar:control",
"payload": {"action": action, "data": payload}
})
return f"Dispatched AR control command: {action} with payload {payload}"
class StableFast3DNotProvisionedError(Exception):
pass
async def is_exe_sidecar_reachable() -> bool:
"""Mock check if EXE sidecar is reachable."""
return True
async def get_inference_target() -> str:
if await is_exe_sidecar_reachable():
return "exe_local_gpu"
else:
return "hf_space_cpu_fallback"
async def find_known_image(description: str) -> str | None:
"""Tier 1: Check SQLite then ChromaDB for a known prior 3D generation."""
# Fast path: SQLite lookup
from backend.memory import database
row = await database.lookup_model_description(description)
if row:
return row["model_path"]
# Semantic match: ChromaDB lookup
from backend.memory.episodic_memory import EpisodicMemory
em = EpisodicMemory("storage/chroma_db")
similar = await em.query_similar_model(description, threshold=0.85)
if similar:
return similar["model_path"]
return None
async def search_web_for_image(description: str) -> str | None:
"""Tier 2: Search the web (Wikimedia Commons) for a matching image."""
import logging
import urllib.parse
import urllib.request
import json
import tempfile
import asyncio
logging.info(f"Tier 2: Searching web for image matching: {description}")
def _search():
# Using Wikipedia/Wikimedia API to find an open-source image
url = f"https://en.wikipedia.org/w/api.php?action=query&prop=pageimages&format=json&piprop=original&titles={urllib.parse.quote(description)}"
req = urllib.request.Request(url, headers={'User-Agent': 'JARVIS_OMEGA/1.0'})
try:
with urllib.request.urlopen(req, timeout=5) as resp:
data = json.loads(resp.read().decode('utf-8'))
pages = data.get("query", {}).get("pages", {})
for page_id, page_data in pages.items():
if "original" in page_data:
img_url = page_data["original"]["source"]
tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
urllib.request.urlretrieve(img_url, tmp.name)
return tmp.name
except Exception as e:
logging.error(f"Web image search failed: {e}")
return None
return await asyncio.to_thread(_search)
async def pollinations_generate_image(description: str) -> str:
"""Tier 3: Generating AI image as ultimate fallback."""
import logging
import urllib.parse
import urllib.request
import tempfile
import asyncio
logging.info(f"Tier 3: Generating AI image for: {description}")
def _generate():
# Fallback AI image generator (using free Pollinations API as proxy for Gemini output generation)
# Note: In production this would use Gemini/Imagen via Vertex AI.
prompt = urllib.parse.quote(f"A high quality realistic 3D asset of {description}, isolated on white background, studio lighting")
url = f"https://image.pollinations.ai/prompt/{prompt}?width=1024&height=1024&nologo=true"
try:
tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
urllib.request.urlretrieve(url, tmp.name)
return tmp.name
except Exception as e:
logging.error(f"AI image generation failed: {e}")
raise e
return await asyncio.to_thread(_generate)
async def log_model_generation(description: str, image_tier: str, model_path: str, persona: str, inference_target: str):
from backend.db import mongodb
from backend.memory import database
from backend.memory.episodic_memory import EpisodicMemory
em = EpisodicMemory("storage/chroma_db")
import asyncio
await asyncio.gather(
mongodb.log_model_generation_mongo(description, image_tier, model_path, persona, inference_target),
em.embed_model_description(description, model_path, persona),
database.insert_model_generation(description, image_tier, model_path, persona, inference_target)
)
async def push_model_to_ar_scene(model_path: str, description: str):
try:
from phone import ws_scene_bus
patch = {
"type": "patch",
"patch": {
"action": "spawn-generated-glb",
"model_url": f"/static/models3d/{model_path.split('/')[-1]}" if '/' in model_path else f"/static/models3d/{model_path}",
"model_name": description
}
}
# S4: broadcast_to_clients is sync — awaiting it raised TypeError and the
# spawn broadcast never reached the AR clients. Use the async variant.
await ws_scene_bus.broadcast_to_clients_async(patch)
except Exception as e:
logging.error(f"Failed to push model to AR scene bus: {e}")
async def spawn_3d_model(description: str, persona: str, use_cloud_fallback: bool = False) -> str:
from backend.services.connectors import stable_fast_3d_local
health = await stable_fast_3d_local.health_check()
if not health["ready"]:
raise StableFast3DNotProvisionedError("Stable Fast 3D model is not installed. Run scripts/provision_stable_fast_3d.sh first.")
image_source_tier = "known"
image = await find_known_image(description)
if not image:
image_source_tier = "web_search"
image = await search_web_for_image(description)
if not image:
image_source_tier = "ai_generated"
image = await pollinations_generate_image(description)
model_path = await stable_fast_3d_local.image_to_3d_local(image, use_cloud_fallback)
if model_path == "INTERACTIVE_PROMPT_REQUIRED":
return "STATUS: LOCAL_MODEL_MISSING. Please ask the user: 'The local SF3D engine is not installed. Should I initiate the 10GB installation script, or route this through the Cloud API for now?' If they choose Cloud API, call this tool again with use_cloud_fallback=True."
await push_model_to_ar_scene(model_path, description)
# S4: determine_inference_target is synchronous — awaiting it raised TypeError
# after every successful spawn, killing the Mongo/Chroma/SQLite logging below.
target = stable_fast_3d_local.determine_inference_target()
await log_model_generation(description, image_source_tier, model_path, persona, target)
return model_path
|