feat: implement file watcher and reviewer

Author: mwfarb

control_plane/main.py

@@ -2,16 +2,20 @@
 import os
 
 # Import all pages to register their routes
-from pages import dashboard, clients, capture, agent, rerun_view, settings
+from pages import anomalies, clients, logs, agent, rerun_view, settings
 
 # Initialize pages
-dashboard.create_page()
+anomalies.create_page()
 clients.create_page()
-capture.create_page()
+logs.create_page()
 agent.create_page()
 rerun_view.create_page()
 settings.create_page()
 
+@ui.page('/')
+def index():
+    ui.navigate.to('/anomalies')
+
 if __name__ in {"__main__", "__mp_main__"}:
     # Read port from config.json and start NiceGUI on config['port'] + 1 (default 8081).
     full_config = settings.load_config()

control_plane/pages/agent.py

@@ -3,6 +3,7 @@
 import subprocess
 import os
 import sys
+from pages.anomalies import fetch_files, get_anomaly_path
 
 def launch_rerun():
     ui.notify('Launching Rerun native viewer...', type='info')
@@ -29,7 +30,11 @@ def agent_page():
             with ui.row().classes('w-full gap-6'):
                 with ui.column().classes('w-1/3'):
                     ui.label('Investigate Issue').classes('text-xl font-bold mb-4')
-                    ui.select(['Latency > 500ms (Camera Rig A)', 'Connection Dropped (Mobile Client B)'], label='Target Anomaly').classes('w-full mb-4')
+                    
+                    anomaly_files = fetch_files(get_anomaly_path())
+                    options = [f"{f['name']} ({f['mtime_str']})" for f in anomaly_files] if anomaly_files else ['No anomalies found']
+                    
+                    ui.select(options, label='Target Anomaly', value=options[0] if options else None).classes('w-full mb-4')
                     ui.textarea(label='Context / Prompt', value='Analyze the provided QUIC videos and MQTT logs to determine why the latency spiked over 500ms on Rig A.').classes('w-full h-32 mb-4')
                     ui.button('Execute Investigation', icon='science', on_click=lambda: ui.notify('Agent investigation dispatched! Context sent to agentic frameworks.')).classes('w-full bg-blue-600')
 

control_plane/pages/anomalies.py

@@ -0,0 +1,108 @@
+from nicegui import app, ui
+import os
+import time
+from pathlib import Path
+from theme import menu
+from pages.settings import load_config
+
+def get_anomaly_path():
+    full_config = load_config()
+    cp_config = full_config.get('control_plane', {})
+    anomaly_path = cp_config.get('logPaths', {}).get('anomalyLogs', 'logs/anomalies/')
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', anomaly_path))
+
+def fetch_files(base_path):
+    files_data = []
+    if not os.path.exists(base_path):
+        return files_data
+        
+    for p in Path(base_path).rglob('*'):
+        if p.is_file():
+            stat = p.stat()
+            files_data.append({
+                'path': str(p),
+                'rel_path': str(p.relative_to(base_path)),
+                'name': p.name,
+                'ext': p.suffix.lower(),
+                'size': stat.st_size,
+                'mtime': stat.st_mtime,
+                'mtime_str': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(stat.st_mtime))
+            })
+    return sorted(files_data, key=lambda x: x['mtime'], reverse=True)
+
+def render_file_content(f):
+    ext = f['ext']
+    path = f['path']
+    rel_path = f['rel_path']
+    
+    # Text-based
+    if ext in ['.txt', '.json', '.yaml', '.yml', '.csv', '.md', '.log']:
+        try:
+            with open(path, 'r', encoding='utf-8') as file:
+                content = file.read()
+                # Truncate if too huge
+                if len(content) > 50000:
+                    content = content[:50000] + "\n...[truncated]"
+                ui.textarea(value=content).props('readonly').classes('w-full h-64 font-mono text-sm bg-gray-50 border p-2')
+        except Exception as e:
+            ui.label(f"Error reading text file: {e}").classes('text-red-500')
+            
+    # Media: Video
+    elif ext in ['.mp4', '.webm', '.ogg']:
+        # To serve this media, it needs an app.add_media_files route.
+        # We can dynamically add a route for the anomaly dir if not already added.
+        ui.video(f'/anomaly_media/{rel_path}').classes('w-full max-w-2xl bg-black rounded')
+        
+    # Media: Image
+    elif ext in ['.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp']:
+        ui.image(f'/anomaly_media/{rel_path}').classes('w-full max-w-2xl rounded')
+        
+    # Audio
+    elif ext in ['.mp3', '.wav']:
+        ui.audio(f'/anomaly_media/{rel_path}').classes('w-full')
+        
+    else:
+        ui.label(f"Binary or unsupported format. File size: {f['size']} bytes.").classes('text-gray-500 italic')
+        ui.button('Download', on_click=lambda p=path: ui.download(p)).classes('mt-2')
+
+def create_page():
+    # Ensure media router is set up once
+    base_path = get_anomaly_path()
+    os.makedirs(base_path, exist_ok=True)
+    app.add_media_files('/anomaly_media', base_path)
+
+    @ui.page('/anomalies')
+    def anomalies_page():
+        # Store state locally to this page connection
+        page_state = {'last_hash': None}
+        
+        with menu('System Anomalies'):
+            ui.label('Detected Anomaly Files').classes('text-xl font-semibold mb-4 text-gray-800')
+            
+            container = ui.column().classes('w-full gap-2')
+            
+            def update_ui():
+                current_files = fetch_files(base_path)
+                
+                # Simple hash/check to avoid re-rendering if no changes
+                state_hash = hash(str([(f['path'], f['mtime'], f['size']) for f in current_files]))
+                if page_state['last_hash'] == state_hash:
+                    return
+                page_state['last_hash'] = state_hash
+                
+                container.clear()
+                with container:
+                    if not current_files:
+                        ui.label(f'No anomalies detected. Watching: {base_path}').classes('text-gray-500 italic p-4')
+                        return
+                        
+                    for f in current_files:
+                        with ui.expansion(f"{f['name']} ({f['mtime_str']})", icon='warning').classes('w-full bg-white border rounded shadow-sm'):
+                            ui.label(f"Path: {f['rel_path']} | Size: {f['size']/1024:.1f} KB").classes('text-xs text-gray-400 mb-2')
+                            render_file_content(f)
+            
+            # Initial render
+            update_ui()
+            
+            # Poll every 3 seconds for file changes
+            ui.timer(3.0, update_ui)

control_plane/pages/capture.py

@@ -1,41 +1,44 @@
-from nicegui import ui
+from nicegui import app, ui
 import aiohttp
+import asyncio
+import json
 from theme import menu, get_local_ip
 from pages.settings import load_config
 
 def create_page():
+    config = load_config()
+    edge_port = config.get('mobile_client', {}).get('port', 8080)
+
     @ui.page('/clients')
     def clients_page():
-        config = load_config()
-        edge_port = config.get('port', 8080)
-        
-        with menu('Mobile Clients & Pairing'):
+        with menu('Mobile Client Connections'):
             ip = get_local_ip()
             server_url = f"ws://{ip}:{edge_port}"
 
             with ui.row().classes('w-full gap-8'):
                 with ui.card().classes('items-center p-6 w-1/3'):
-                    ui.label('Server Pairing').classes('text-xl font-bold mb-4')
+                    ui.label('Server Pairing').classes('text-xl font-bold mb-4 text-gray-800')
                     ui.label('Scan this QR code with the AllSpark app:').classes('text-sm text-gray-600 text-center mb-4')
 
                     # Generate a QR code using a public API for the prototype
                     ui.image(f'https://api.qrserver.com/v1/create-qr-code/?size=150x150&data={server_url}').classes('w-32 h-32')
                     ui.label(server_url).classes('mt-4 font-mono font-bold bg-gray-100 p-2 rounded max-w-full break-all text-center')
 
-                with ui.column().classes('w-2/3'):
-                    ui.label('Active Connections').classes('text-xl font-bold mb-4')
+                with ui.column().classes('w-2/3 gap-4'):
+                    ui.label('Active Connections').classes('text-xl font-bold text-gray-800')
 
-                    cards_container = ui.column().classes('w-full')
+                    cards_container = ui.column().classes('w-full gap-2')
 
                     async def fetch_and_render_clients():
                         try:
                             async with aiohttp.ClientSession() as session:
                                 async with session.get(f'http://127.0.0.1:{edge_port}/api/status', timeout=2) as resp:
                                     if resp.status == 200:
                                         data = await resp.json()
-                                        clients_list = data.get('clients', [])
-                                        # Also handle `connectedClients` just in case
-                                        if not clients_list and 'connectedClients' in data:
+                                        clients_list = data.get('connections', [])
+                                        if not clients_list and 'clients' in data:
+                                            clients_list = data['clients']
+                                        elif not clients_list and 'connectedClients' in data:
                                             clients_list = data['connectedClients']
                                         render_clients(clients_list)
                                     else:
@@ -49,7 +52,7 @@ async def request_upload(client_id):
                                 payload = {"command": "upload"}
                                 async with session.post(f'http://127.0.0.1:{edge_port}/api/command/{client_id}', json=payload) as resp:
                                     if resp.status == 200:
-                                        ui.notify(f'Upload command sent to {client_id}!', type='positive')
+                                        ui.notify(f'Manual upload command sent to {client_id}!', type='positive')
                                     else:
                                         ui.notify(f'Failed to send command: HTTP {resp.status}', type='negative')
                         except Exception as e:
@@ -59,7 +62,7 @@ def render_clients(clients_data):
                         cards_container.clear()
                         with cards_container:
                             if clients_data is None:
-                                ui.label(f'Edge server offline or /api/status not reachable on port {edge_port}.').classes('text-red-500 italic p-4')
+                                ui.label(f'Edge server offline or api/status unreachable on port {edge_port}.').classes('text-red-500 italic p-4')
                                 return
                             if not clients_data:
                                 ui.label('Waiting for mobile rig connections...').classes('text-gray-500 italic p-4')
@@ -68,17 +71,18 @@ def render_clients(clients_data):
                             for c in clients_data:
                                 client_id = c.get('id', 'Unknown')
                                 c_type = c.get('type', 'Rig')
-                                with ui.card().classes('w-full mb-2'):
+                                with ui.card().classes('w-full border shadow-sm'):
                                     with ui.row().classes('w-full justify-between items-center'):
                                         ui.label(f'{c_type} (ID: {client_id})').classes('font-bold')
                                         ui.badge('Online', color='green')
-                                    # Depending on how edge API shapes IP/Stats, display it safely
                                     ui.label(f"Stats: {c.get('details', 'No details available')}").classes('text-sm text-gray-600 mt-1')
+                                    
+                                    if c.get("lastFilename"):
+                                        ui.label(f"Latest File: {c['lastFilename']}").classes('text-sm text-green-700 mt-1')
+                                        
                                     with ui.row().classes('mt-4 items-center gap-2'):
-                                        ui.input('Start Time', value='2026-03-25T09:00:00').props('type=datetime-local borderless dense').classes('w-48')
-                                        ui.input('End Time', value='2026-03-25T10:00:00').props('type=datetime-local borderless dense').classes('w-48')
-                                        ui.button('Request Upload', on_click=lambda ci=client_id: request_upload(ci)).classes('bg-blue-600 text-white')
+                                        ui.button('Force Sync Uploads', on_click=lambda ci=client_id: request_upload(ci)).classes('bg-blue-600 text-white')
 
-                    # Poll the API
-                    ui.timer(5.0, fetch_and_render_clients)
+                    # Poll the API for client list 
+                    ui.timer(2.0, fetch_and_render_clients)
                     ui.timer(0.1, fetch_and_render_clients, once=True)