Update app.py

app.py

@@ -1,235 +1,150 @@
-import numpy as np
-import matplotlib.pyplot as plt
-import io, base64
-from fastapi import FastAPI
-from fastapi.responses import HTMLResponse
-from sklearn.decomposition import PCA
-from sklearn.cluster import AgglomerativeClustering
-from sklearn.metrics.pairwise import euclidean_distances
 import time
-
-app = FastAPI()
-
-class AdaptiveVectorSystem:
-    def _calculate_score(self, target, constituents):
-        """
-        Generates a score (-1.0 to 10.0) representing convergence 'effort'.
-        """
-        if len(constituents) == 0:
-            return -1.0
-            
-        # Calculate distances from the calculated center to the points used
-        dists = np.linalg.norm(constituents - target, axis=1)
-        
-        # Mean distance (how far usually)
-        mean_dist = np.mean(dists)
-        # Standard Deviation (how chaotic/scattered)
-        std_dev = np.std(dists)
-        
-        # Heuristic: We want a high score for Low Mean and Low Std Dev.
-        # We normalize based on the mean_dist itself to make it scale-invariant.
-        # If std_dev is high relative to mean_dist, score drops.
+import collections
+import threading
+from flask import Flask, jsonify, request
+from flask_cors import CORS
+
+app = Flask(__name__)
+CORS(app)
+
+class SimEngine:
+    def __init__(self):
+        self.nodes = {}       
+        self.cells =[]       
+        self.buffer = collections.deque()
+        self.running = False
         
-        variation_coefficient = (std_dev / (mean_dist + 1e-9))
+        # Dial Dashboard State
+        self.mode = 'inference'       # 'inference' (free A&B) or 'training' (clamped A&B)
+        self.distribution = 'uniform' # 'uniform' (50/50 split) or 'individual' (vertex k-values)
+        self.problem_type = 'add'     # 'add' or 'mult'
+        self.asymmetric = False       # dampen retroactive pushes to prevent exploding values
         
-        # Base score starts at 10
-        # We penalize for high variation (chaos) and raw distance
-        penalty = (variation_coefficient * 6.0) + (mean_dist * 0.1)
+        self.reset()
+
+    def reset(self):
+        # Initialize semantic 'embeddings' in 3D latent space. X is the logic value for this simple PoC.
+        self.nodes = {
+            'A': {'x': 2.0, 'y': 1.0, 'z': 0.0, 'anchored': False, 'k': 1.0},
+            'B': {'x': 3.0, 'y': -1.0, 'z': 0.0, 'anchored': False, 'k': 0.8}, 
+            'C': {'x': 10.0, 'y': 0.0, 'z': 1.0, 'anchored': True,  'k': 1.0}  
+        }
+        # One mesh cell connects all 3 constraints
+        self.cells =[{'id': 'Cell_1', 'a': 'A', 'b': 'B', 'c': 'C'}]
+        self.buffer.clear()
+        self.logs =[]
+        self.iteration = 0
+
+    def add_log(self, msg):
+        self.logs.insert(0, f"Iter {self.iteration}: {msg}")
+        if len(self.logs) > 50: self.logs.pop() # Keep log buffer clean
+
+    def set_problem(self, target_value):
+        # Target objective is permanently clamped
+        self.nodes['C']['x'] = float(target_value)
+        self.nodes['C']['anchored'] = True
         
-        score = 10.0 - penalty
-        return max(-1.0, min(10.0, score))
-
-    def predict_point(self, vectors, mode='global'):
-        data = np.array(vectors)
-        
-        # --- GLOBAL MODE ---
-        # "Force a fit for everyone." 
-        # Great for converged data, terrible for split data.
-        if mode == 'global':
-            center = np.mean(data, axis=0)
-            score = self._calculate_score(center, data)
-            return center, score, data
-
-        # --- CLUSTER MODE ---
-        # "Find the strongest gravity well."
-        elif mode == 'cluster':
-            # 1. Compute Pairwise Distances to understand the "Scale" of the data
-            dist_matrix = euclidean_distances(data, data)
-            # Flatten matrix and remove zeros (self-distance) to get average spacing
-            all_dists = dist_matrix[np.triu_indices(len(data), k=1)]
-            avg_global_dist = np.mean(all_dists)
-            
-            # 2. DYNAMIC THRESHOLDING
-            # We say: To belong to a group, points must be significantly closer 
-            # than the global average. (e.g., 0.6 * average)
-            dynamic_thresh = avg_global_dist * 0.65 
-            
-            # 3. Cluster with this dynamic threshold
-            clusterer = AgglomerativeClustering(
-                n_clusters=None,
-                metric='euclidean',
-                linkage='ward',
-                distance_threshold=dynamic_thresh
-            )
-            labels = clusterer.fit_predict(data)
-            
-            # 4. Find the "Best" Cluster
-            # We look for the Largest cluster, but we ignore "Noise" (clusters of size 1 or 2)
-            unique_labels, counts = np.unique(labels, return_counts=True)
+        if self.mode == 'training':
+            self.nodes['A']['anchored'] = True
+            self.nodes['B']['anchored'] = True
+            self.add_log(f"Training initiated. C clamped to: {target_value}")
+        else:
+            self.nodes['A']['anchored'] = False
+            self.nodes['B']['anchored'] = False
+            self.add_log(f"Inference initiated. Free mesh calculating towards: {target_value}")
+
+        self.trigger_cells() # Jumpstart event cascade
+
+    def trigger_cells(self):
+        """Cell mathematical constraint check. Pushes structural 'error' to buffer if not solved."""
+        for cell in self.cells:
+            na, nb, nc = self.nodes[cell['a']], self.nodes[cell['b']], self.nodes[cell['c']]
             
-            # Filter out tiny clusters (noise)
-            valid_clusters = [l for l, c in zip(unique_labels, counts) if c > 2]
+            # Using basic arithmetic logic in Dim X for visibility
+            valA, valB, valC = na['x'], nb['x'], nc['x']
+            predictedC = (valA + valB) if self.problem_type == 'add' else (valA * valB)
             
-            if not valid_clusters:
-                # Fallback if everything is noise: treat everything as one group
-                return self.predict_point(data, mode='global')
-                
-            # Pick largest of the valid clusters
-            # (You could also pick the 'densest' here, but largest is usually safest)
-            best_label = max(valid_clusters, key=lambda l: counts[np.where(unique_labels == l)][0])
+            error = predictedC - valC 
             
-            # 5. Extract Data
-            cluster_vectors = data[labels == best_label]
-            center = np.mean(cluster_vectors, axis=0)
-            score = self._calculate_score(center, cluster_vectors)
-            
-            return center, score, cluster_vectors
+            if abs(error) > 0.05: # Yield Limit threshold
+                # Add retroactive tension to A and B
+                self.buffer.append({'target': cell['a'], 'error_vector': error, 'cell': cell})
+                self.buffer.append({'target': cell['b'], 'error_vector': error, 'cell': cell})
 
-# --- VISUALIZATION LOGIC ---
-def generate_plot(mode='global', scenario='split'):
-    # Generate 128-dimension vectors
-    np.random.seed(int(time.time())
-                  ) # Consistent seed for demo
-    
-    if scenario == 'split':
-        # Create two dense islands far apart
-        # Island 1: centered at 0
-        c1 = np.random.normal(0, 0.5, (100, 128))
-        # Island 2: centered at 10 (In 128D, distance approx sqrt(128*100) = ~113 units away)
-        c2 = np.random.normal(8, 0.5, (100, 128))
-        # Noise: Random scatter
-        noise = np.random.uniform(-5, 15, (10, 128))
-        data = np.vstack([c1, c2, noise])
-    else:
-        # One Tight Cluster
-        data = np.random.normal(0, 1.0, (50, 128))
-
-    # Run System
-    sys = AdaptiveVectorSystem()
-    center_vec, score, used_vectors = sys.predict_point(data, mode)
-
-    # PCA for 2D View
-    # Important: Fit PCA on Input + Center so they share the same coordinate space
-    pca = PCA(n_components=2)
-    all_points = np.vstack([data, center_vec])
-    projected = pca.fit_transform(all_points)
-    
-    pts_2d = projected[:-1]
-    center_2d = projected[-1]
-
-    # --- Plotting ---
-    plt.figure(figsize=(7, 5), facecolor='#202020')
-    ax = plt.gca()
-    ax.set_facecolor('#303030')
-    
-    # Logic to identify which points were used (for coloring)
-    # We compare the rows of 'used_vectors' to 'data' to find indices
-    # Note: In production, pass indices around. For demo, we do a quick check.
-    is_used = np.zeros(len(data), dtype=bool)
+    def physics_step(self):
+        if not self.buffer: return False # Queue is empty, logic equilibrium reached.
+        
+        event = self.buffer.popleft()
+        t_id, err = event['target'], event['error_vector']
+        t_node = self.nodes[t_id]
+
+        # Ignore locked objects
+        if t_node['anchored'] and self.mode != 'training':
+            return True 
+
+        # Implement Asymmetry Rule (Stop resonance explosions)
+        direction_damper = 0.3 if self.asymmetric else 1.0
+
+        if self.mode == 'inference':
+            # --- Adapt Topology Geometry ---
+            # Moves to minimize structural stress based on dials
+            base_force = (-err * 0.02 * direction_damper)
+            if self.distribution == 'uniform':
+                t_node['x'] += base_force 
+                t_node['y'] -= base_force * 0.1 # visual spatial twisting
+            elif self.distribution == 'individual':
+                t_node['x'] += base_force * t_node['k']
+            
+        elif self.mode == 'training':
+            # --- Learn Structural Stiffness 'K' ---
+            # Node geometry is locked. System adjusts how elastic the point is.
+            base_grad = abs(err) * 0.005 * direction_damper
+            if self.distribution == 'individual':
+                 t_node['k'] -= base_grad 
+        
+        # Move cascaded, so tell network to re-measure stress.
+        self.trigger_cells() 
+        self.iteration += 1
+        return True
+
+engine = SimEngine()
+
+def run_loop():
+    while True:
+        if engine.running: engine.physics_step()
+        time.sleep(0.015) # Simulated latency/Buffer cycle speed
+
+threading.Thread(target=run_loop, daemon=True).start()
+
+# --------- INTERFACE API ---------
+
+@app.route('/state', methods=['GET'])
+def get_state():
+    return jsonify({
+        'nodes': engine.nodes,
+        'buffer_size': len(engine.buffer),
+        'iteration': engine.iteration,
+        'logs': engine.logs,
+        'mode': engine.mode
+    })
+
+@app.route('/apply_config', methods=['POST'])
+def config():
+    data = request.json
+    engine.running = False # Pause briefly for logic change
     
-    # A quick way to mask used vectors using broadcasting approximation
-    # (Since floats are tricky, we assume exact match from the split)
-    if mode == 'global':
-        is_used[:] = True
-    else:
-        # Brute force match for visualization accuracy
-        for uv in used_vectors:
-            for i, dv in enumerate(data):
-                if np.array_equal(uv, dv):
-                    is_used[i] = True
-                    break
-
-    # 1. Plot IGNORED points (Grey, transparent)
-    if not np.all(is_used):
-        plt.scatter(pts_2d[~is_used, 0], pts_2d[~is_used, 1], 
-                   c='#555555', alpha=0.3, s=30, label='Ignored (Noise/Other)')
-
-    # 2. Plot USED points (Bright Cyan)
-    plt.scatter(pts_2d[is_used, 0], pts_2d[is_used, 1], 
-               c='#00e5ff', alpha=0.8, s=40, edgecolors='none', label='Constituent Inputs')
-
-    # 3. Draw "Gravity Lines" (faint lines from used points to center)
-    # Only draw lines if there aren't too many points, to keep it clean
-    if np.sum(is_used) < 100:
-        for pt in pts_2d[is_used]:
-            plt.plot([pt[0], center_2d[0]], [pt[1], center_2d[1]], 
-                    c='#00e5ff', alpha=0.15, linewidth=1)
-
-    # 4. Plot The PREDICTED POINT (Red X)
-    plt.scatter(center_2d[0], center_2d[1], 
-               c='#ff3366', s=200, marker='X', edgecolors='white', linewidth=1.5, 
-               label='Generated Vector', zorder=10)
-
-    # Styling
-    plt.title(f"Mode: {mode.upper()} | Score: {score:.2f}/10", color='white', fontsize=12, pad=10)
-    plt.grid(True, color='#444444', linestyle='--', alpha=0.5)
+    engine.mode = data.get('mode', engine.mode)
+    engine.distribution = data.get('distribution', engine.distribution)
+    engine.problem_type = data.get('problem_type', engine.problem_type)
+    engine.asymmetric = data.get('asymmetric', engine.asymmetric)
+    target = data.get('target', 10.0)
     
-    # Legend formatting
-    leg = plt.legend(facecolor='#303030', edgecolor='#555555', fontsize=8, loc='best')
-    for text in leg.get_texts():
-        text.set_color("white")
-
-    # Axis colors
-    ax.tick_params(axis='x', colors='white')
-    ax.tick_params(axis='y', colors='white')
-    for spine in ax.spines.values():
-        spine.set_edgecolor('#555555')
-
-    buf = io.BytesIO()
-    plt.savefig(buf, format='png', bbox_inches='tight')
-    plt.close()
-    return base64.b64encode(buf.getvalue()).decode('utf-8')
-
-@app.get("/", response_class=HTMLResponse)
-async def root():
-    img_global = generate_plot('global', 'split')
-    img_cluster = generate_plot('cluster', 'split')
-    img_tight = generate_plot('global', 'tight')
+    engine.set_problem(target)
+    engine.add_log(f"[DIALS CHANGED]: phase={engine.mode}, split={engine.distribution}, type={engine.problem_type}, async={engine.asymmetric}")
     
-    return f"""
-    <html>
-        <body style="font-family: 'Segoe UI', sans-serif; background:#121212; color:#e0e0e0; text-align:center; padding:20px;">
-            <h1 style="margin-bottom:10px;">Vector Convergence System</h1>
-            <p style="color:#888; margin-bottom:40px;">Dynamic Thresholding Algorithm</p>
-            
-            <div style="display:flex; flex-wrap:wrap; justify-content:center; gap:20px;">
-                <!-- SCENARIO A -->
-                <div style="background:#1e1e1e; padding:20px; border-radius:12px; border:1px solid #333;">
-                    <h2 style="color:#aaa; border-bottom:1px solid #333; padding-bottom:10px;">Scenario: Split Data</h2>
-                    <div style="display:flex; gap:20px;">
-                        <div>
-                            <h3 style="color:#00e5ff;">Global Mode</h3>
-                            <div style="font-size:0.8em; color:#888; margin-bottom:5px;">Averages everything (Score -1 to 2)</div>
-                            <img src="data:image/png;base64,{img_global}" width="400" style="border-radius:8px;"/>
-                        </div>
-                        <div>
-                            <h3 style="color:#ff3366;">Cluster Mode (Revised)</h3>
-                            <div style="font-size:0.8em; color:#888; margin-bottom:5px;">Identifies largest mass (Score 8 to 10)</div>
-                            <img src="data:image/png;base64,{img_cluster}" width="400" style="border-radius:8px;"/>
-                        </div>
-                    </div>
-                </div>
-
-                <!-- SCENARIO B -->
-                <div style="background:#1e1e1e; padding:20px; border-radius:12px; border:1px solid #333;">
-                    <h2 style="color:#aaa; border-bottom:1px solid #333; padding-bottom:10px;">Scenario: Converged Data</h2>
-                    <div>
-                        <h3 style="color:#00e5ff;">Global Mode</h3>
-                         <div style="font-size:0.8em; color:#888; margin-bottom:5px;">Efficient calculation (Score ~10)</div>
-                        <img src="data:image/png;base64,{img_tight}" width="400" style="border-radius:8px;"/>
-                    </div>
-                </div>
-            </div>
-        </body>
-    </html>
-    """
+    engine.running = data.get('is_running', False)
+    return jsonify(success=True)
+
+if __name__ == '__main__':
+    print("Semantic Latent Topography Core started on :5000.")
+    app.run(port=7860, debug=True, use_reloader=False)