Upload qads/core/system.py

qads/core/system.py

@@ -0,0 +1,133 @@
+"""QADS Main System Integration."""
+import numpy as np
+from typing import Dict, Any, Optional, Tuple, List
+from .config import QADSConfig
+from ..quantum.core import QuantumDecisionCore
+from ..planner.hybrid import HybridPlanner
+from ..simulation.environment import SimulationEnvironment
+from ..perception.fusion import SensorFusion
+from ..control.interface import ControlInterface
+
+
+class QADSSystem:
+    """
+    Quantum Autonomous Decision System.
+    
+    Integrates perception, quantum decision core, hybrid planning,
+    reinforcement learning, and control into a unified framework.
+    """
+    
+    def __init__(self, config: Optional[QADSConfig] = None):
+        self.config = config or QADSConfig()
+        self.perception = SensorFusion(self.config.perception)
+        self.quantum_core = QuantumDecisionCore(self.config.quantum)
+        self.planner = HybridPlanner(self.config.planner, self.quantum_core)
+        self.control = ControlInterface()
+        self.env: Optional[SimulationEnvironment] = None
+        self.state_history: List[Dict[str, Any]] = []
+        self.mission_log: List[Dict[str, Any]] = []
+        
+    def initialize_environment(self, env_config: Optional[Dict[str, Any]] = None) -> SimulationEnvironment:
+        """Initialize simulation environment."""
+        config = env_config or {}
+        self.env = SimulationEnvironment(self.config.simulation, **config)
+        return self.env
+    
+    def perceive(self, raw_sensors: Dict[str, np.ndarray]) -> Dict[str, Any]:
+        """Process raw sensor data into structured world state."""
+        return self.perception.process(raw_sensors)
+    
+    def plan(self, 
+             start: Tuple[float, ...], 
+             goal: Tuple[float, ...],
+             world_state: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Generate plan from start to goal."""
+        if world_state is None and self.env is not None:
+            world_state = self.env.get_world_state()
+            
+        plan = self.planner.plan(start, goal, world_state)
+        self.state_history.append({
+            'start': start,
+            'goal': goal,
+            'plan': plan,
+            'world_state': world_state
+        })
+        return plan
+    
+    def execute(self, plan: Dict[str, Any]) -> Dict[str, Any]:
+        """Execute plan and return execution results."""
+        if self.env is None:
+            raise RuntimeError("Environment not initialized")
+            
+        trajectory = []
+        rewards = []
+        done = False
+        step = 0
+        
+        current_pos = plan['start']
+        actions = plan.get('actions', [])
+        
+        for action in actions:
+            if done:
+                break
+                
+            next_state, reward, terminated, truncated, info = self.env.step(action)
+            trajectory.append({
+                'position': next_state['position'],
+                'action': action,
+                'reward': reward,
+                'info': info
+            })
+            rewards.append(reward)
+            done = terminated or truncated
+            step += 1
+            
+        result = {
+            'success': plan['goal_reached'] if 'goal_reached' in plan else False,
+            'trajectory': trajectory,
+            'total_reward': sum(rewards),
+            'steps': step,
+            'plan_metrics': plan.get('metrics', {})
+        }
+        self.mission_log.append(result)
+        return result
+    
+    def run_mission(self, 
+                    start: Tuple[float, ...], 
+                    goal: Tuple[float, ...],
+                    max_retries: int = 3) -> Dict[str, Any]:
+        """Run complete mission: perceive → plan → execute."""
+        for attempt in range(max_retries):
+            world_state = self.env.get_world_state() if self.env else None
+            plan = self.plan(start, goal, world_state)
+            result = self.execute(plan)
+            
+            if result['success']:
+                result['attempts'] = attempt + 1
+                return result
+                
+            # Replan if failed
+            if attempt < max_retries - 1:
+                self.planner.update_world_state(self.env.get_world_state())
+                
+        result['attempts'] = max_retries
+        return result
+    
+    def get_quantum_metrics(self) -> Dict[str, float]:
+        """Get quantum computation metrics."""
+        return self.quantum_core.get_metrics()
+    
+    def get_mission_summary(self) -> Dict[str, Any]:
+        """Get summary of all missions."""
+        if not self.mission_log:
+            return {}
+            
+        successes = sum(1 for m in self.mission_log if m['success'])
+        return {
+            'total_missions': len(self.mission_log),
+            'successes': successes,
+            'failures': len(self.mission_log) - successes,
+            'success_rate': successes / len(self.mission_log),
+            'avg_steps': np.mean([m['steps'] for m in self.mission_log]),
+            'avg_reward': np.mean([m['total_reward'] for m in self.mission_log])
+        }