Spaces:
Runtime error
Runtime error
File size: 3,419 Bytes
edae06c | 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 | import os
import ast
import requests
import time
import json
# Configuration
PROJECT_ROOT = "c:/Users/Nauti/Desktop/LOGOS CURSOR/logos"
# We'll use the ingest endpoint for the manifold update
API_ENDPOINT = "http://127.0.0.1:5000/ingest"
def parse_code_dna(file_path):
"""
Extracts the 'Skeleton' of the code: Classes, Functions, and Imports.
This creates the 'Gold Threads' (connections) in the Visualizer.
"""
try:
with open(file_path, "r", encoding="utf-8") as source:
tree = ast.parse(source.read())
except:
return None # Skip broken files
dna = {"classes": [], "functions": [], "imports": []}
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
dna["classes"].append(node.name)
elif isinstance(node, ast.FunctionDef):
dna["functions"].append(node.name)
elif isinstance(node, ast.Import):
for alias in node.names:
dna["imports"].append(alias.name)
elif isinstance(node, ast.ImportFrom):
dna["imports"].append(node.module or "")
return dna
def index_system():
print(f"[GEMMA] Initiating Full-Spectrum Sweep of: {PROJECT_ROOT}...")
file_count = 0
for root, _, files in os.walk(PROJECT_ROOT):
for file in files:
if file.endswith(".py"):
full_path = os.path.join(root, file)
# 1. Extract DNA
dna = parse_code_dna(full_path)
if not dna: continue
# 2. Calculate "Semantic Mass" (Complexity)
# Weighted score based on class/func density
mass = len(dna['classes']) * 10 + len(dna['functions']) * 2
# 3. Construct Swarm Packet
# Target move: add <mass>
packet = {
"source": "GEMMA",
"value": mass,
"type": "code_indexing",
"content": f"Index module {file} (Mass: {mass})",
"meta": {
"filename": file,
"path": root,
"dna": dna,
"mass": mass
}
}
# 4. Pulse to Manifold
print(f"[INDEX] {file} -> Mass: {mass} | Imports: {len(dna['imports'])}")
try:
# We use the existing ingest endpoint to update the manifold
requests.post(API_ENDPOINT, json=packet)
# Also move the swarm cursor for each file
# We simulate this by sending an 'add <mass>' command to the neural logic
# This uses the FAST-PATH we just patched!
requests.post("http://127.0.0.1:5000/v1/chat/completions", json={
"messages": [{"role": "user", "content": f"add {mass}"}]
})
except Exception as e:
print(f"[ERROR] Pulse failed for {file}: {e}")
file_count += 1
time.sleep(0.1) # Smooth telemetry gradient
print(f"\n[GEMMA] Index Complete. {file_count} Modules Mapped to Manifold.")
if __name__ == "__main__":
index_system()
|