physics2.py - master - Lasith/RocketEngineAI

physics2.py
324 строки12 KB
Lasith
upload files
21 апр 2025, 19:13
21 апр 2025, 19:130f9763e
import sqlite3
import json
import random
import time
import threading
from datetime import datetime
from math import exp

class KnowledgeBase:
    def __init__(self, db_name='thrust_concepts.db'):
        self.conn = sqlite3.connect(db_name, check_same_thread=False)
        self.lock = threading.Lock()
        self.create_tables()

    def create_tables(self):
        with self.lock:
            cursor = self.conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS concepts (
                    id INTEGER PRIMARY KEY,
                    principle TEXT,
                    parameters TEXT,
                    created_at DATETIME
                )''')
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS experiments (
                    id INTEGER PRIMARY KEY,
                    concept_id INTEGER,
                    result TEXT,
                    anomaly_score REAL,
                    created_at DATETIME,
                    FOREIGN KEY(concept_id) REFERENCES concepts(id)
                )''')
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS laws (
                    id INTEGER PRIMARY KEY,
                    description TEXT,
                    formula TEXT,
                    discovered_at DATETIME
                )''')
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_concepts ON concepts(principle)')
            self.conn.commit()

    def save_concept(self, principle, parameters):
        with self.lock:
            try:
                cursor = self.conn.cursor()
                cursor.execute('''
                    INSERT INTO concepts (principle, parameters, created_at)
                    VALUES (?, ?, ?)
                ''', (principle, json.dumps(parameters), datetime.now()))
                self.conn.commit()
                return cursor.lastrowid
            except sqlite3.Error as e:
                print(f"Database error: {e}")
                return None

    def save_experiment(self, concept_id, result, anomaly_score):
        with self.lock:
            try:
                cursor = self.conn.cursor()
                cursor.execute('''
                    INSERT INTO experiments (concept_id, result, anomaly_score, created_at)
                    VALUES (?, ?, ?, ?)
                ''', (concept_id, json.dumps(result), anomaly_score, datetime.now()))
                self.conn.commit()
                return cursor.lastrowid
            except sqlite3.Error as e:
                print(f"Database error: {e}")
                return None

    def save_law(self, description, formula):
        with self.lock:
            try:
                cursor = self.conn.cursor()
                cursor.execute('''
                    SELECT 1 FROM laws 
                    WHERE description = ? AND formula = ?
                ''', (description, formula))
                if not cursor.fetchone():
                    cursor.execute('''
                        INSERT INTO laws (description, formula, discovered_at)
                        VALUES (?, ?, ?)
                    ''', (description, formula, datetime.now()))
                    self.conn.commit()
                    return cursor.lastrowid
                return None
            except sqlite3.Error as e:
                print(f"Database error: {e}")
                return None

    def get_all_concepts(self):
        with self.lock:
            cursor = self.conn.cursor()
            cursor.execute('SELECT * FROM concepts')
            return cursor.fetchall()

    def get_all_experiments(self):
        with self.lock:
            cursor = self.conn.cursor()
            cursor.execute('SELECT * FROM experiments')
            return cursor.fetchall()

    def get_all_laws(self):
        with self.lock:
            cursor = self.conn.cursor()
            cursor.execute('SELECT * FROM laws')
            return cursor.fetchall()

    def close(self):
        self.conn.close()

class PhysicsEngine:
    def __init__(self):
        self.base_constants = {
            'c': 299792458,
            'G': 6.67430e-11,
            'epsilon_0': 8.854187817e-12
        }

    def calculate_thrust(self, principle, params):
        try:
            if principle == 'quantum_vacuum_fluctuation':
                return params['energy_density'] * params['modulation_frequency'] / self.base_constants['c']
            
            elif principle == 'spacetime_curvature':
                if params['focus_distance'] < 1e-15:
                    return 0
                return (params['mass_energy'] * self.base_constants['G'] 
                        / (params['focus_distance']**2))
            
            elif principle == 'dark_matter_interaction':
                return params['interaction_cross_section'] * (params['field_strength']**2)
            
            elif principle == 'magnetic_monopole_acceleration':
                return (params['monopole_density'] 
                       * params['acceleration_field'] 
                       * self.base_constants['epsilon_0'])
            
            elif principle == 'hawking_radiation_harvesting':
                return (params['black_hole_mass'] 
                       * params['harvesting_efficiency'] 
                       * 1e3)
            
            return params.get('base_thrust', 0) * random.uniform(0.5, 1.5)
        
        except KeyError as e:
            print(f"Missing required parameter: {e}")
            return 0
        except ZeroDivisionError:
            print("Division by zero prevented")
            return 0

    def detect_anomalies(self, result):
        anomaly_score = 0
        
        if result.get('achieved_velocity', 0) > self.base_constants['c']:
            anomaly_score += (result['achieved_velocity'] / self.base_constants['c'] - 1) * 10
        
        if result.get('efficiency', 0) > 1.0:
            anomaly_score += (result['efficiency'] - 1) * 5
        
        if result.get('thrust', 0) > 1e20:
            anomaly_score += 5
            
        return anomaly_score

class ResearchSystem:
    def __init__(self):
        self.kb = KnowledgeBase()
        self.physics = PhysicsEngine()
        self.running = False
        self.thread = None
        
        self.thrust_principles = [
            'quantum_vacuum_fluctuation',
            'spacetime_curvature',
            'dark_matter_interaction',
            'magnetic_monopole_acceleration',
            'hawking_radiation_harvesting'
        ]
        
        self.mandatory_params = {
            'quantum_vacuum_fluctuation': ['energy_density', 'modulation_frequency'],
            'spacetime_curvature': ['mass_energy', 'focus_distance'],
            'dark_matter_interaction': ['interaction_cross_section', 'field_strength'],
            'magnetic_monopole_acceleration': ['monopole_density', 'acceleration_field'],
            'hawking_radiation_harvesting': ['black_hole_mass', 'harvesting_efficiency']
        }
        
        self.parameter_ranges = {
            'energy_density': (1e3, 1e18),
            'modulation_frequency': (1e3, 1e12),
            'mass_energy': (1e6, 1e30),
            'focus_distance': (1e-9, 1e3),
            'interaction_cross_section': (1e-30, 1e-18),
            'field_strength': (1e3, 1e12),
            'monopole_density': (1e5, 1e20),
            'acceleration_field': (1e-5, 1e5),
            'black_hole_mass': (1e20, 1e35),
            'harvesting_efficiency': (0.01, 0.99)
        }

    def generate_concept(self):
        principle = random.choice(self.thrust_principles)
        params = {}
        
        for param in self.mandatory_params[principle]:
            params[param] = random.uniform(*self.parameter_ranges[param])
        
        for param in self.parameter_ranges:
            if param not in self.mandatory_params[principle] and random.random() > 0.3:
                params[param] = random.uniform(*self.parameter_ranges[param])
        
        return principle, params

    def run_experiment(self, concept):
        principle, params = concept
        result = {}
        
        try:
            result['thrust'] = self.physics.calculate_thrust(principle, params)
            result['energy_consumption'] = params.get('energy_density', 0) * random.uniform(0.8, 1.2)
            result['efficiency'] = (result['thrust'] / result['energy_consumption'] 
                                   if result['energy_consumption'] != 0 else 0)
            result['achieved_velocity'] = result['thrust'] * random.uniform(1e3, 1e5)
            
            anomaly_score = self.physics.detect_anomalies(result)
            
            if anomaly_score > 15:
                new_law = self._generate_new_law(result)
                if new_law:
                    self.kb.save_law(new_law['description'], new_law['formula'])
            
            return result, anomaly_score
        
        except Exception as e:
            print(f"Experiment failed: {e}")
            return {}, 0

    def _generate_new_law(self, result):
        components = [
            f"T^{random.choice(['2', '3', '1/2'])}",
            f"exp(-E/{max(result.get('energy_consumption', 1e3), 1e3):.1e})",
            f"ln({random.choice(['v', 't', 'm'])})"
        ]
        formula = " * ".join(random.sample(components, 2))
        return {
            'description': f"New {random.choice(['Quantum', 'Relativistic', 'Dark'])} Dynamics Law",
            'formula': f"{formula} + {random.uniform(0.1, 0.9):.2f}"
        }

    def start_research(self):
        self.running = True
        while self.running:
            principle, params = self.generate_concept()
            concept_id = self.kb.save_concept(principle, params)
            if concept_id:
                result, anomaly_score = self.run_experiment((principle, params))
                if result:
                    self.kb.save_experiment(concept_id, result, anomaly_score)
            time.sleep(0.5)

    def stop_research(self):
        self.running = False
        if self.thread:
            self.thread.join()

    def print_statistics(self):
        concepts = self.kb.get_all_concepts()
        experiments = self.kb.get_all_experiments()
        laws = self.kb.get_all_laws()

        print(f"\nFound concepts: {len(concepts)}")
        print(f"Experiments conducted: {len(experiments)}")
        print(f"Discovered laws: {len(laws)}\n")

        if concepts:
            print("Recent concepts:")
            for c in concepts[-3:]:
                params = json.loads(c[2])
                formatted = ", ".join(
                    f"{k}: {v:.2e}" if isinstance(v, float) else f"{k}: {v}"
                    for k, v in params.items()
                )
                print(f"[{c[3]}] {c[1]}: {formatted}")

        if laws:
            print("\nRecent laws:")
            for law in laws[-3:]:
                print(f"{law[1]} ({law[3]})\nFormula: {law[2]}")

class Interface:
    def __init__(self):
        self.research = ResearchSystem()
        
    def run(self):
        print("Advanced Space Propulsion Research System")
        try:
            while True:
                cmd = input("\nCommands: start/stop/stats/exit > ").lower()
                
                if cmd == 'start' and not self.research.running:
                    self.research.thread = threading.Thread(target=self.research.start_research)
                    self.research.thread.start()
                    print("Research started...")
                    
                elif cmd == 'stop' and self.research.running:
                    self.research.stop_research()
                    print("Research stopped")
                    
                elif cmd == 'stats':
                    self.research.print_statistics()
                    
                elif cmd == 'exit':
                    self.research.stop_research()
                    print("Exiting system...")
                    break
        finally:
            self.research.kb.close()

if __name__ == "__main__":
    interface = Interface()
    interface.run()