adding more function to shka and pytagoras

req.txt

@@ -8,5 +8,6 @@ scipy
 scikit-learn
 pandas
 psutil
-
+python-dotenv
+beautifulsoup4
 

satellites/atlasSat/atlas.py

@@ -6,6 +6,7 @@ from satellites.base_satellite import VegapunkSatellite
 import logging
 import subprocess
 import time
+from utils.logger import get_logger
 
 
 role = "Sécurité et Automatisation"
@@ -16,8 +17,8 @@ class Atlas(VegapunkSatellite):
         self.monitored_directories = []
         self.email_config ={}
         self.external_systems = {}
-        logging.basicConfig(filename='atlas.log', level=logging.INFO)
-
+        # logging.basicConfig(filename='atlas.log', level=logging.INFO)
+        self.logger = get_logger("atlas")
     def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
         task_type = task.get("type")
         if task_type == "monitor_directory":

satellites/base_satellite.py

@@ -1,7 +1,7 @@
-
-
 from abc import ABC, abstractmethod
 from typing import  Dict, Any
+import logging
+
 
 
 class VegapunkSatellite(ABC):

satellites/lilithSat/lilith.py

@@ -1,7 +1,7 @@
 from satellites.base_satellite import VegapunkSatellite
 from typing import Dict, Any, List
-import logging
 import random
+from utils.logger import get_logger
 
 role = " Explorer des solution non conventionnel , creatove,voir risqué"
 
@@ -14,9 +14,9 @@ class Lilith(VegapunkSatellite):
         self.idea_categories = ["Technologie", "Art", "Science", "Société", "Environnement"]
         self.innovation_levels = ["Incrémentale", "Radicale", "Disruptive"]
         self.unconventional_approaches = ["Pensée inversée", "Analogies lointaines", "Combinaison aléatoire","Contraintes extrêmes"]
-        logging.basicConfig(filename='lilith_log.txt', level=logging.INFO)
+        # logging.basicConfig(filename='lilith_log.txt', level=logging.INFO)
         self.external_apis = {}
-
+        self.logger = get_logger("lilith")
     def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
         task_type = task.get("type")
         if task_type == "generate_idea":

satellites/pythagorasSat/pythagoras.py

@@ -6,8 +6,20 @@ import matplotlib.pyplot as plt
 import io
 import base64
 import random
+from utils.logger import get_logger
+import pandas as pd
+from bs4 import BeautifulSoup
+import json
+import os
+from dotenv import load_dotenv
+import requests
 import logging
 
+load_dotenv()
+
+
+
+
 
 
 role = "mathématiques, statistiques et analyse de données"
@@ -16,6 +28,13 @@ fonction = "Effectuer des calculs complexes et analyser des ensembles de donnée
 class Pythagoras(VegapunkSatellite):
     def __init__(self):
         super().__init__(name="Pythagoras", specialty="Role")
+        self.llm_api_key = os.getenv("LLM_API_KEY")
+        self.llm_api_url = "https://api.openai.com/v1/chat/completions"  # Example using OpenAI's API
+        self.research_databases = {
+            "scientific": "https://api.example-scientific-db.com/search",
+            "news": "https://api.example-news-db.com/search",
+            "general": "https://api.example-general-db.com/search"
+        }
         self.mathematical_constants = {
             "pi": np.pi,
             "e": np.e,
@@ -23,6 +42,7 @@ class Pythagoras(VegapunkSatellite):
         }
         self.resources = {}
         self.external_apis = {}
+        self.logger = get_logger("pythagoras")
 
     def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
         task_type = task.get('type')
@@ -32,8 +52,17 @@ class Pythagoras(VegapunkSatellite):
             return self._analyze_dataset(task.get('data'))
         elif task_type == "statistical_test":
             return self._perform_statistical_test(task.get('test_type'), task.get('data'))
+        if task_type == "analyze_data":
+            result = self.analyze_data_2(task["data"])
+        elif task_type == "conduct_research":
+            result = self.conduct_research(task["topic"], task.get("depth", "medium"))
+        elif task_type == "extract_information":
+            result = self.extract_information(task["content"], task.get("keywords", []))
         else:
-            return {"error": "Tâche non reconnue"}
+            result = f"Tâche non reconnue : {task_type}"
+
+        self.log_activity(f"Tâche traitée : {task_type}, Résultat : {result}")
+        return {"result": result}
 
     def _perform_calculation(self, operation: str, values: List[float]) -> Dict[str, Any]:
         if not operation or not values:
@@ -111,6 +140,92 @@ class Pythagoras(VegapunkSatellite):
         except Exception as e:
             return {"error": f"Erreur lors du test statistique: {str(e)}"}
 
+    def analyze_data_2(self, data: List[Dict[str, Any]]) -> Dict[str, Any]:
+        df = pd.DataFrame(data)
+        analysis = {
+            "summary": df.describe().to_dict(),
+            "correlations": df.corr().to_dict(),
+            "trends": self._detect_trends(df),
+            "outliers": self._detect_outliers(df)
+        }
+        return analysis
+
+    def conduct_research(self, topic: str, depth: str = "medium") -> Dict[str, Any]:
+        research_results = {}
+        for db_name, db_url in self.research_databases.items():
+            research_results[db_name] = self._search_database(db_url, topic, depth)
+
+        summary = self._summarize_research(topic, research_results)
+        return {
+            "topic": topic,
+            "depth": depth,
+            "results": research_results,
+            "summary": summary
+        }
+
+    def extract_information(self, content: str, keywords: List[str] = []) -> Dict[str, Any]:
+        prompt = f"Extraire les informations clés du texte suivant, en se concentrant sur les mots-clés {keywords} si fournis : {content}"
+        extracted_info = self._query_llm(prompt)
+
+        return {
+            "original_content_length": len(content),
+            "extracted_information": extracted_info,
+            "keywords_used": keywords
+        }
+
+    def _detect_trends(self, df: pd.DataFrame) -> Dict[str, Any]:
+        trends = {}
+        for column in df.select_dtypes(include=[np.number]).columns:
+            trend = stats.linregress(range(len(df)), df[column])
+            trends[column] = {
+                "slope": trend.slope,
+                "intercept": trend.intercept,
+                "r_value": trend.rvalue,
+                "p_value": trend.pvalue,
+                "trend": "increasing" if trend.slope > 0 else "decreasing"
+            }
+        return trends
+
+    def _detect_outliers(self, df: pd.DataFrame) -> Dict[str, List[Any]]:
+        outliers = {}
+        for column in df.select_dtypes(include=[np.number]).columns:
+            z_scores = np.abs(stats.zscore(df[column]))
+            outliers[column] = df[column][z_scores > 3].tolist()
+        return outliers
+
+    def _search_database(self, db_url: str, topic: str, depth: str) -> List[Dict[str, Any]]:
+        # Simuler une recherche dans une base de données externe
+        # Dans une implémentation réelle, cela ferait un appel API à la base de données
+        return [
+            {"title": f"Résultat 1 pour {topic}", "summary": f"Résumé du résultat 1 pour {topic}"},
+            {"title": f"Résultat 2 pour {topic}", "summary": f"Résumé du résultat 2 pour {topic}"}
+        ]
+
+    def _summarize_research(self, topic: str, research_results: Dict[str, List[Dict[str, Any]]]) -> str:
+        # Utiliser le LLM pour résumer les résultats de recherche
+        research_summary = json.dumps(research_results)
+        prompt = f"Résumez les résultats de recherche suivants sur le sujet '{topic}' : {research_summary}"
+        return self._query_llm(prompt)
+
+    def _query_llm(self, prompt: str) -> str:
+        headers = {
+            "Authorization": f"Bearer {self.llm_api_key}",
+            "Content-Type": "application/json"
+        }
+        data = {
+            "model": "gpt-3.5-turbo",
+            "messages": [{"role": "user", "content": prompt}]
+        }
+        try:
+            response = requests.post(self.llm_api_url, headers=headers, json=data)
+            response.raise_for_status()
+            return response.json()['choices'][0]['message']['content']
+        except requests.RequestException as e:
+            return f"Erreur lors de la requête LLM : {str(e)}"
+
+    def log_activity(self, activity: str):
+        logging.info(activity)
+
     def communicate_with_stellar(self, message: Dict[str, Any]) -> Dict[str, Any]:
         print(f"{self.name} envoie un message à Stellar: {message}")
         return {"Statut": "Message reçu", "message": "Stellar a bien reçu le message de Pythagoras"}

satellites/shakaSat/shaka.py

@@ -4,7 +4,14 @@ import nltk
 from nltk.tokenize import word_tokenize
 from nltk.corpus import stopwords
 import string
+from utils.logger import get_logger
+import requests
+import os
+import json
 import logging
+from dotenv import load_dotenv
+
+load_dotenv()
 
 role = "logique, éthique et analyse"
 fonction = "Vérifier la cohérence des informations et filtrer les informations non éthiques"
@@ -13,7 +20,19 @@ class Shaka(VegapunkSatellite):
     def __init__(self):
         super().__init__(name="Shaka", specialty=role)
         self.nlp = self._initialize_nlp_tools()
-        logging.basicConfig(filename='shaka.log', level=logging.INFO)
+        # logging.basicConfig(filename='shaka.log', level=logging.INFO)
+        self.llm_api_key = os.getenv("LLM_API_KEY")
+        self.llm_api_url = "https://api.openai.com/v1/chat/completions"  # Example using OpenAI's API
+        self.fact_check_api = "https://factchecktools.googleapis.com/v1alpha1/claims:search"
+        self.fact_check_api_key = os.getenv("FACT_CHECK_API_KEY")
+        self.ethical_filter_active = True
+        self.ethical_guidelines = [
+            "Ne pas promouvoir la violence ou des activités illégales",
+            "Éviter les discriminations basées sur la race, le genre, la religion, etc.",
+            "Respecter la vie privée et les données personnelles",
+            "Promouvoir l'honnêteté et la transparence",
+            "Éviter la désinformation et les fake news"]
+        self.logger = get_logger("shaka")
 
     def _initialize_nlp_tools(self):
         resources = ["stopwords", "punctuation", "averaged_perceptron_tagger", "wordnet", "punkt_tab"]
@@ -36,8 +55,19 @@ class Shaka(VegapunkSatellite):
             return self._analyse_text(task['content'])
         elif task_type == 'raisonnement_logique':
             return self._logical_reasoning(task['hypothese'], task['question'])
+        elif task_type == "ethical_check":
+            result = self.check_ethics(task["content"])
+        elif task_type == "fact_check":
+            result = self.verify_facts(task["claim"])
+        elif task_type == "provide_recommendations":
+            result = self.provide_ethical_recommendations(task["content"])
+        elif task_type == "toggle_ethical_filter":
+            result = self.toggle_ethical_filter()
         else:
-            return {"error": "Tâche non reconnue"}
+            result = f"Tâche non reconnue : {task_type}"
+
+        self.log_activity(f"Tâche traitée : {task_type}, Résultat : {result}")
+        return {"result": result}
 
     def _analyse_text(self, texte: str) -> Dict[str, Any]:
         try:
@@ -66,6 +96,73 @@ class Shaka(VegapunkSatellite):
                           ("pertinentes" if relevance_score > 0.5 else "peu pertinentes") + " à la question."
         }
 
+    def check_ethics(self, content: str) -> Dict[str, Any]:
+        if not self.ethical_filter_active:
+            return {"status": "Le filtre éthique est désactivé", "content": content}
+
+        prompt = f"Analysez le contenu suivant et déterminez s'il est éthique selon ces directives : {self.ethical_guidelines}. Contenu : '{content}'"
+        response = self._query_llm(prompt)
+        is_ethical = "éthique" in response.lower() and "non éthique" not in response.lower()
+
+        return {
+            "content": content,
+            "is_ethical": is_ethical,
+            "analysis": response
+        }
+
+    def verify_facts(self, claim: str) -> Dict[str, Any]:
+        # Utilisation de l'API Google Fact Check Tools (nécessite une clé API valide)
+        params = {
+            "key": self.fact_check_api_key,
+            "query": claim
+        }
+        try:
+            response = requests.get(self.fact_check_api, params=params)
+            response.raise_for_status()
+            fact_checks = response.json().get("claims", [])
+            if fact_checks:
+                return {
+                    "claim": claim,
+                    "fact_check_result": fact_checks[0].get("claimReview", [])[0].get("textualRating", "Inconnu"),
+                    "source": fact_checks[0].get("claimReview", [])[0].get("publisher", {}).get("name", "Inconnu")
+                }
+            else:
+                return {"claim": claim, "fact_check_result": "Aucune vérification trouvée"}
+        except requests.RequestException as e:
+            return {"claim": claim, "error": f"Erreur lors de la vérification des faits : {str(e)}"}
+
+    def provide_ethical_recommendations(self, content: str) -> Dict[str, Any]:
+        prompt = f"Le contenu suivant a été jugé non éthique : '{content}'. Proposez une version alternative qui respecte ces directives éthiques : {self.ethical_guidelines}"
+        response = self._query_llm(prompt)
+        return {
+            "original_content": content,
+            "ethical_recommendation": response
+        }
+
+    def toggle_ethical_filter(self) -> Dict[str, Any]:
+        self.ethical_filter_active = not self.ethical_filter_active
+        status = "activé" if self.ethical_filter_active else "désactivé"
+        return {"message": f"Le filtre éthique est maintenant {status}"}
+
+    def _query_llm(self, prompt: str) -> str:
+        headers = {
+            "Authorization": f"Bearer {self.llm_api_key}",
+            "Content-Type": "application/json"
+        }
+        data = {
+            "model": "gpt-3.5-turbo",
+            "messages": [{"role": "user", "content": prompt}]
+        }
+        try:
+            response = requests.post(self.llm_api_url, headers=headers, json=data)
+            response.raise_for_status()
+            return response.json()['choices'][0]['message']['content']
+        except requests.RequestException as e:
+            return f"Erreur lors de la requête LLM : {str(e)}"
+
+    def log_activity(self, activity: str):
+        logging.info(activity)
+
     def communicate_with_stellar(self, message: Dict[str, Any]) -> Dict[str, Any]:
         print(f"{self.name} envoie un message à Stellar: {message}")
         return {"Statut": "Message reçu", "message": "Stellar a bien reçu le message"}

satellites/yorkSat/york.py

@@ -2,6 +2,7 @@ import psutil
 import time
 from typing import Dict, Any, List
 from satellites.base_satellite import VegapunkSatellite
+from utils.logger import get_logger
 import logging
 
 role = "Gestion des ressources et maintenance système"
@@ -19,8 +20,9 @@ class York(VegapunkSatellite):
             "network": 75,
         }
         self.maintenance_schedule = {}
-        logging.basicConfig(filename='york.log', level=logging.INFO)
+        # logging.basicConfig(filename='york.log', level=logging.INFO)
         self.external_apis = {}
+        from utils.logger import get_logger
 
     def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
         task_type = task.get("type")

tests/shaka_test.py

@@ -21,6 +21,17 @@ def test_shaka():
 
     logic_result = shaka.process_task(test_logic)
     print("Résultat de la tâche de logique:", logic_result)
+    # Ajout de tâches à la file d'attente
+    shaka.add_task({"type": "ethical_check",
+                    "content": "Les gens devraient toujours dire la vérité, même si cela peut blesser quelqu'un."})
+    shaka.add_task({"type": "fact_check", "claim": "La Terre est plate."})
+    shaka.add_task({"type": "provide_recommendations", "content": "Tous les immigrants devraient être expulsés."})
+    shaka.add_task({"type": "toggle_ethical_filter"})
+
+    # Traitement des tâches
+    while task := shaka.get_next_task():
+        result = shaka.process_task(task)
+        print(f"Résultat de la tâche : {result}")
 
     print("Connexion au Punkrecord")
     shaka.update_from_punkrecord()