Dijital İkiz Oluşturun

import json
from datetime import datetime
from openai import OpenAI

client = OpenAI()

class DigitalTwin:
    """A digital twin that models a physical system."""
    
    def __init__(self, name, system_type, properties):
        self.name = name
        self.system_type = system_type
        self.properties = properties
        self.state_history = []
        self.current_state = properties.copy()
        self.alerts = []
    
    def update_state(self, sensor_data):
        """Update the twin with real-world sensor data."""
        timestamp = datetime.now().isoformat()
        self.state_history.append({
            "timestamp": timestamp,
            "state": self.current_state.copy(),
            "sensor_data": sensor_data
        })
        
        # Update current state
        for key, value in sensor_data.items():
            if key in self.current_state:
                self.current_state[key] = value
        
        # Check for anomalies
        self._check_anomalies(sensor_data)
    
    def _check_anomalies(self, data):
        """Use AI to detect anomalies in sensor data."""
        if len(self.state_history) < 5:
            return
        
        recent = self.state_history[-5:]
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"""You monitor a 
{self.system_type} digital twin named '{self.name}'.
Normal operating ranges: {json.dumps(self.properties)}

Analyze recent sensor readings for anomalies.
Return JSON: {{"anomaly_detected": true/false, 
"severity": "low/medium/high",
"details": "explanation",
"recommended_action": "what to do"}}"""},
                {"role": "user", "content": json.dumps(recent)}
            ],
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        if result.get("anomaly_detected"):
            self.alerts.append(result)

# Create a digital twin of a manufacturing machine
machine_twin = DigitalTwin(
    name="CNC Machine A-01",
    system_type="CNC Milling Machine",
    properties={
        "temperature": 45.0,
        "vibration": 0.3,
        "spindle_speed": 8000,
        "power_consumption": 15.5,
        "tool_wear": 0.0
    }
)

# Simulate sensor updates
machine_twin.update_state({
    "temperature": 47.2,
    "vibration": 0.32,
    "spindle_speed": 7950,
    "power_consumption": 15.8,
    "tool_wear": 0.05
})

class SimulationEngine:
    """Run what-if simulations on the digital twin."""
    
    def __init__(self, digital_twin):
        self.twin = digital_twin
        self.client = OpenAI()
    
    def run_scenario(self, scenario_description, steps=10):
        """Simulate a scenario and predict outcomes."""
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"""You are simulating a 
{self.twin.system_type}. Current state:
{json.dumps(self.twin.current_state)}

History of last readings:
{json.dumps(self.twin.state_history[-5:] if self.twin.state_history else [])}

Simulate {steps} time steps for the given scenario. 
For each step, predict all property values realistically.

Return JSON:
{{"scenario": "description",
 "steps": [{{"step": 1, "time": "T+1h", 
   "state": {{"prop": value}}, 
   "events": "what happens"}}],
 "final_outcome": "summary",
 "risk_assessment": "low/medium/high",
 "recommendations": ["action 1"]}}"""},
                {"role": "user", "content": scenario_description}
            ],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

# Run a what-if simulation
simulator = SimulationEngine(machine_twin)
result = simulator.run_scenario(
    "What happens if the cooling system fails and temperature "
    "rises by 2 degrees per hour?",
    steps=8
)

for step in result["steps"]:
    print(f"Step {step['step']} ({step['time']}): "
          f"Temp={step['state'].get('temperature', 'N/A')} "
          f"- {step['events']}")
print(f"\nOutcome: {result['final_outcome']}")
print(f"Risk: {result['risk_assessment']}")

class PredictiveMaintenanceEngine:
    """Predict maintenance needs using the digital twin."""
    
    def __init__(self, digital_twin):
        self.twin = digital_twin
        self.client = OpenAI()
    
    def predict_maintenance(self):
        """Analyze trends and predict when maintenance is needed."""
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"""You are a predictive 
maintenance AI for a {self.twin.system_type}.

Current state: {json.dumps(self.twin.current_state)}
History: {json.dumps(self.twin.state_history[-20:])}
Known alerts: {json.dumps(self.twin.alerts)}

Predict maintenance needs. Return JSON:
{{"components": [
  {{"name": "component", 
   "health": 0-100,
   "estimated_failure": "time estimate",
   "maintenance_type": "preventive/corrective",
   "priority": "low/medium/high/critical",
   "description": "what needs to be done"}}
],
"next_maintenance_window": "recommended time",
"estimated_downtime": "hours",
"cost_estimate": "relative (low/medium/high)"}}"""},
                {"role": "user", "content": "Generate maintenance predictions based on current data."}
            ],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

# Run predictions
maintenance = PredictiveMaintenanceEngine(machine_twin)
predictions = maintenance.predict_maintenance()

for comp in predictions["components"]:
    print(f"{comp['name']}: Health {comp['health']}% "
          f"- {comp['priority']} priority"
          f"\n  {comp['description']}")