Upload 10 files

diagnostics.py

@@ -1,20 +1,24 @@
 
-import platform
-import psutil
-import shutil
+import traceback
+from datetime import datetime
 
-def run_diagnostics():
-    report = {
-        "platform": platform.platform(),
-        "cpu_percent": psutil.cpu_percent(),
-        "memory": psutil.virtual_memory()._asdict(),
-        "disk": shutil.disk_usage("/"),
-        "status": "🟢 All systems functional"
-    }
+class DiagnosticEngine:
+    def __init__(self):
+        self.error_logs = []
 
-    if report["memory"]["percent"] > 90:
-        report["status"] = "🟡 High memory usage"
-    if report["cpu_percent"] > 85:
-        report["status"] = "🟡 High CPU load"
+    def capture_exception(self, e, context=""):
+        tb = traceback.format_exc()
+        error_entry = {
+            "timestamp": datetime.utcnow().isoformat(),
+            "error": str(e),
+            "traceback": tb,
+            "context": context
+        }
+        self.error_logs.append(error_entry)
+        return error_entry
 
-    return report
+    def get_logs(self):
+        return self.error_logs
+
+    def last_error(self):
+        return self.error_logs[-1] if self.error_logs else None

external_api_integrations.py

@@ -0,0 +1,35 @@
+
+import requests
+
+# Define endpoints for demonstration
+API_ENDPOINTS = {
+    "weather": "https://api.open-meteo.com/v1/forecast?latitude=52.52&longitude=13.41&current_weather=true",
+    "news": "https://newsapi.org/v2/top-headlines?country=us&apiKey=YOUR_NEWS_API_KEY",
+    "stocks": "https://finnhub.io/api/v1/quote?symbol=AAPL&token=YOUR_STOCK_API_KEY"
+}
+
+def get_weather():
+    try:
+        response = requests.get(API_ENDPOINTS["weather"])
+        return response.json()
+    except Exception as e:
+        return {"error": str(e)}
+
+def get_news():
+    try:
+        response = requests.get(API_ENDPOINTS["news"])
+        return response.json()
+    except Exception as e:
+        return {"error": str(e)}
+
+def get_stock():
+    try:
+        response = requests.get(API_ENDPOINTS["stocks"])
+        return response.json()
+    except Exception as e:
+        return {"error": str(e)}
+
+if __name__ == "__main__":
+    print("🌦️ Weather:", get_weather())
+    print("🗞️ News:", get_news())
+    print("📈 Stock:", get_stock())

model_switcher.py

@@ -0,0 +1,26 @@
+
+import os
+from sentence_transformers import SentenceTransformer
+
+class ModelManager:
+    def __init__(self, default_model="all-MiniLM-L6-v2"):
+        self.current_model_name = default_model
+        self.model = SentenceTransformer(default_model)
+        self.available_models = [
+            "all-MiniLM-L6-v2",
+            "paraphrase-MiniLM-L3-v2",
+            "multi-qa-MiniLM-L6-cos-v1"
+        ]
+
+    def switch_model(self, model_name):
+        if model_name not in self.available_models:
+            raise ValueError(f"Model '{model_name}' is not in the list of available models.")
+        self.current_model_name = model_name
+        self.model = SentenceTransformer(model_name)
+        return f"Switched to model: {model_name}"
+
+    def get_current_model(self):
+        return self.current_model_name
+
+    def list_models(self):
+        return self.available_models

model_updater.py

@@ -0,0 +1,38 @@
+
+import json
+import os
+import time
+from datetime import datetime
+
+MODEL_VERSION_PATH = "model_version.json"
+
+def load_version():
+    if os.path.exists(MODEL_VERSION_PATH):
+        with open(MODEL_VERSION_PATH, "r") as f:
+            return json.load(f)
+    return {"version": "1.0.0", "last_updated": None}
+
+def update_model_version(new_version: str):
+    version_info = {
+        "version": new_version,
+        "last_updated": datetime.utcnow().isoformat()
+    }
+    with open(MODEL_VERSION_PATH, "w") as f:
+        json.dump(version_info, f, indent=2)
+    print(f"✅ Model updated to version {new_version}")
+
+def auto_update():
+    version_data = load_version()
+    current_version = version_data["version"]
+    print(f"🔍 Current model version: {current_version}")
+
+    # Simulate version check and update
+    major, minor, patch = map(int, current_version.split('.'))
+    patch += 1
+    new_version = f"{major}.{minor}.{patch}"
+    update_model_version(new_version)
+
+if __name__ == "__main__":
+    while True:
+        auto_update()
+        time.sleep(3600)  # Update every hour

planner_scheduler.py

@@ -0,0 +1,36 @@
+
+import datetime
+from typing import List, Dict, Optional
+
+class Task:
+    def __init__(self, description: str, deadline: Optional[str] = None):
+        self.description = description
+        self.deadline = deadline
+        self.created_at = datetime.datetime.now()
+        self.status = "scheduled"
+
+    def to_dict(self) -> Dict:
+        return {
+            "description": self.description,
+            "deadline": self.deadline,
+            "created_at": self.created_at.isoformat(),
+            "status": self.status
+        }
+
+class PlannerScheduler:
+    def __init__(self):
+        self.tasks: List[Task] = []
+
+    def schedule_task(self, description: str, deadline: Optional[str] = None) -> str:
+        task = Task(description, deadline)
+        self.tasks.append(task)
+        return f"Task '{description}' scheduled."
+
+    def list_tasks(self, status_filter: Optional[str] = None) -> List[Dict]:
+        return [t.to_dict() for t in self.tasks if status_filter is None or t.status == status_filter]
+
+    def update_task_status(self, index: int, new_status: str) -> str:
+        if 0 <= index < len(self.tasks):
+            self.tasks[index].status = new_status
+            return f"Task '{self.tasks[index].description}' marked as {new_status}."
+        return "Invalid task index."

prompt_optimizer.py

@@ -0,0 +1,29 @@
+
+import random
+
+class PromptOptimizer:
+    def __init__(self):
+        self.prompts = [
+            "What is your main objective today?",
+            "How can I assist with your current task?",
+            "What’s the next priority to focus on?",
+        ]
+        self.performance_data = {}
+
+    def get_prompt(self):
+        return random.choice(self.prompts)
+
+    def record_feedback(self, prompt, success=True):
+        if prompt not in self.performance_data:
+            self.performance_data[prompt] = {"success": 0, "fail": 0}
+        if success:
+            self.performance_data[prompt]["success"] += 1
+        else:
+            self.performance_data[prompt]["fail"] += 1
+
+    def optimize_prompts(self):
+        # Basic strategy: remove consistently failing prompts
+        self.prompts = [
+            p for p in self.prompts
+            if self.performance_data.get(p, {}).get("fail", 0) < 3
+        ]

secure_api_gateway.py

@@ -0,0 +1,23 @@
+
+from fastapi import FastAPI, Request, Header, HTTPException
+
+app = FastAPI()
+
+AUTHORIZED_KEYS = {
+    "admin-key-123": "admin",
+    "dev-key-456": "developer",
+    "user-key-789": "user"
+}
+
+@app.middleware("http")
+async def verify_api_key(request: Request, call_next):
+    api_key = request.headers.get("x-api-key")
+    if not api_key or api_key not in AUTHORIZED_KEYS:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+    response = await call_next(request)
+    return response
+
+@app.get("/secure-data")
+async def secure_data(x_api_key: str = Header(...)):
+    user_role = AUTHORIZED_KEYS.get(x_api_key, "unknown")
+    return {"status": "success", "role": user_role, "data": "🔐 confidential data"}

self_assessment.py

@@ -0,0 +1,36 @@
+
+import time
+import json
+from datetime import datetime
+
+class SelfAssessmentLogger:
+    def __init__(self, file_path="self_assessment_log.json"):
+        self.file_path = file_path
+        self.log = []
+
+    def log_assessment(self, module_name, status, details=""):
+        entry = {
+            "timestamp": datetime.utcnow().isoformat(),
+            "module": module_name,
+            "status": status,
+            "details": details
+        }
+        self.log.append(entry)
+        self._persist()
+
+    def assess_module(self, module_name, test_func):
+        try:
+            result = test_func()
+            status = "PASS" if result else "FAIL"
+            self.log_assessment(module_name, status)
+            return status
+        except Exception as e:
+            self.log_assessment(module_name, "ERROR", str(e))
+            return "ERROR"
+
+    def _persist(self):
+        with open(self.file_path, "w") as f:
+            json.dump(self.log, f, indent=2)
+
+    def get_log(self):
+        return self.log

self_evaluator.py

@@ -0,0 +1,31 @@
+
+import time
+import logging
+
+logging.basicConfig(level=logging.INFO)
+
+class SelfEvaluator:
+    def __init__(self):
+        self.evaluation_log = []
+
+    def evaluate(self, task_output: str):
+        quality_score = self.heuristic_score(task_output)
+        self.evaluation_log.append({
+            "output": task_output,
+            "score": quality_score,
+            "timestamp": time.time()
+        })
+        logging.info(f"Self-evaluation completed: Score={quality_score}")
+        return quality_score
+
+    def heuristic_score(self, output: str):
+        # Simple evaluation heuristic based on length and keyword presence
+        if "error" in output.lower():
+            return 0.2
+        elif len(output) < 20:
+            return 0.5
+        else:
+            return 0.9
+
+    def get_log(self):
+        return self.evaluation_log

user_feedback.py

@@ -0,0 +1,29 @@
+
+from fastapi import FastAPI, Request
+from pydantic import BaseModel
+from typing import List
+import datetime
+
+app = FastAPI()
+
+feedback_storage = []
+
+class FeedbackEntry(BaseModel):
+    user_id: str
+    feedback_text: str
+    rating: int  # Scale 1–5
+
+@app.post("/submit_feedback")
+async def submit_feedback(entry: FeedbackEntry):
+    timestamped_entry = {
+        "user_id": entry.user_id,
+        "feedback": entry.feedback_text,
+        "rating": entry.rating,
+        "timestamp": datetime.datetime.utcnow().isoformat()
+    }
+    feedback_storage.append(timestamped_entry)
+    return {"status": "success", "received": timestamped_entry}
+
+@app.get("/feedback_log")
+def get_feedback_log() -> List[dict]:
+    return feedback_storage