ajoute et implementation de la communication etre Satellite

main.py

@@ -0,0 +1,32 @@
+from stellar.base_stellar import Stellar
+from satellites.atlasSat.atlas import Atlas
+from satellites.lilithSat.lilith import Lilith
+from satellites.pythagorasSat.pythagoras import Pythagoras
+from satellites.shakaSat.shaka import Shaka
+from satellites.edisonSat.edison import Edison
+from satellites.yorkSat.york import York
+from typing import Dict, Any
+
+stellar = Stellar()
+atlas = Atlas()
+lilith = Lilith()
+pythagoras = Pythagoras()
+shaka = Shaka()
+edison = Edison()
+york = York()
+
+
+
+stellar.register_satellites(atlas)
+stellar.register_satellites(lilith)
+stellar.register_satellites(pythagoras)
+stellar.register_satellites(shaka)
+stellar.register_satellites(edison)
+stellar.register_satellites(york)
+
+response = Stellar.route_communication("Atlas", "York", {"type": "check_resources"})
+
+results = Stellar.broadcast_message("Atlas", {"type": "check_resources"})
+
+
+print(results)

req.txt

@@ -10,4 +10,5 @@ pandas
 psutil
 python-dotenv
 beautifulsoup4
+langchain-anthropic
 

satellites/atlasSat/atlas.py

@@ -123,4 +123,33 @@ class Atlas(VegapunkSatellite):
         status.update({
             "Monitored_directories": self.monitored_directories,
             "Email_config": self.email_config
-        })
+        })
+
+
+
+    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
+
+        if message.get("type") == "task":
+            task_result = self.process_task(message.get("task"))
+            return {"status": "Traitement effectué", "result": task_result}
+        elif message.get("type") == "email_config":
+            self.email_config = message.get("content")
+            return {"status": "Configuration email mise à jour", "result": self.email_config}
+        elif message.get("type") == "system_command":
+            command_result = self.execute_system_command(message.get("command"))
+            return {"status": "Commande système exécutée", "result": command_result}
+        elif message.get("type") == "file_management":
+            file_result = self.manage_files(message.get("action"), message.get("file_path"))
+            return {"status": "Gestion de fichier effectuée", "result": file_result}
+        elif message.get("type") == "monitor_directory":
+            monitor_result = self.monitor_directory(message.get("directory"))
+            return {"status": "Surveillance de répertoire effectuée", "result": monitor_result}
+        elif message.get("type") == "email":
+            email_result = self.send_email(message.get("to"), message.get("subject"), message.get("body"))
+            return {"status": "Email envoye", "result": email_result}
+        elif message.get("type") == "status_report":
+            result = self.report_status()
+            return {"status": "Rapport de status généré", "result": result}
+        elif message.get("type") == "check_changes":
+            changes = self.check_directory_changes()
+            return {"status": "Changements vérifiés", "result": changes}

satellites/base_satellite.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import  Dict, Any
+from typing import  Dict, Any,Optional
 import logging
 
 
@@ -60,12 +60,35 @@ class VegapunkSatellite(ABC):
         # Methode pour mettre a jour de la base de connaissance local du satellite depuis punkrecord
         pass
 
-    # def communicate_with_other_satellite(self, satellite: VegapunkSatellite, message: Dict[str, Any]) -> Dict[str, Any]:
-    #     # Methode pour communiquer avec un autre satellite
-    #     pass
+        # Methode pour communiquer avec un autre satellite
+    def communicate_with_other_satellite(self, satellite: 'VegapunkSatellite', message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        if not isinstance(satellite, VegapunkSatellite):
+            logging.error(f"Le satellite spécifié n'est pas valide :{type(satellite)}")
+            return None
+
+        try:
+            response = satellite.receive_communication(self.name,message)
+            logging.info(f"Communication avec {satellite.name} : {response}")
+            return response
+        except Exception as e:
+            logging.error(f"Erreur lors de la communication avec {satellite.name} : {str(e)}")
+            return None
+
+    def receive_communication(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        # Methode pour recevoir une communication d'un autre satellite
+        logging.info(f"{self.name} a reçu un message : {message}")
+        return {"status": "Message reçu","from":{self.name}, "details": message}
+
+    @abstractmethod
+    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any])->Dict[str,Any]:
+        """
+        traite le message recu d'un autre satellite
+        a implementer dans chaque classe de satellite specifique
+        """
+        pass
+
+
 
-#
-#
 # class Satellite:
 #     def __init__(self, name, specialty):
 #         self.name = name

satellites/lilithSat/lilith.py

@@ -2,6 +2,7 @@ from satellites.base_satellite import VegapunkSatellite
 from typing import Dict, Any, List
 import random
 from utils.logger import get_logger
+import logging
 
 role = " Explorer des solution non conventionnel , creatove,voir risqué"
 
@@ -114,3 +115,23 @@ class Lilith(VegapunkSatellite):
         # Ici, vous pourriez implémenter la logique pour mettre à jour les approches créatives ou les domaines d'innovation
 
 
+    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
+        if message.get("type") == "task":
+            task_result = self.process_task(message.get("task"))
+            return {"status": "Traitement effectué", "result": task_result}
+        elif message.get("type") == "generate_idea":
+            idea_result = self.generate_creative_idea(message.get("domain"))
+            return {"status": "Idée générée", "result": idea_result}
+        elif message.get("type") == "solve_problem":
+            solution_result = self.propose_unconventional_solution(message.get("problem"))
+            return {"status": "Solution proposée", "result": solution_result}
+        elif message.get("type") == "brainstorm":
+            brainstorm_result = self.conduct_brainstorming_session(message.get("topic"), message.get("duration"))
+            return {"status": "Session de brainstorming effectuée", "result": brainstorm_result}
+        elif message.get("type") == "challenge_assumption":
+            assumption_result = self.challenge_assumption(message.get("assumption"))
+            return {"status": "Assomption challengée", "result": assumption_result}
+        elif message.get("type") == "status_report":
+            result = self.report_status()
+            return {"status": "Rapport de status généré", "result": result}
+

satellites/pythagorasSat/pythagoras.py

@@ -12,14 +12,18 @@ from bs4 import BeautifulSoup
 import json
 import os
 from dotenv import load_dotenv
+from openai import OpenAI
 import requests
 import logging
 
 load_dotenv()
 
 
+# client = OpenAI(os.getenv("LLM_API_KEY"))
 
-
+client =OpenAI(
+    organization="LLM_API_KEY",
+)
 
 
 role = "mathématiques, statistiques et analyse de données"
@@ -241,4 +245,25 @@ class Pythagoras(VegapunkSatellite):
             "Constantes_mathématiques": list(self.mathematical_constants.keys()),
             "Opérations_disponibles": ["mean", "median", "std_dev", "correlation"]
         })
-        return status
+        return status
+
+
+    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
+        if message.get("type")== "task":
+            task_result = self.process_task(message.get("task"))
+            return {"status": "Traitement effectué", "result": task_result}
+        elif message.get("type") == "research":
+            research_result = self.conduct_research(message.get("topic"), message.get("depth"))
+            return {"status": "Recherche effectuée", "result": research_result}
+        elif message.get("type") == "information_extraction":
+            info_result = self.extract_information(message.get("content"), message.get("keywords", []))
+            return {"status": "Extraction d'information effectuée", "result": info_result}
+        elif message.get("type") == "update_constants":
+            self.mathematical_constants.update(message.get("constants", {}))
+            return {"status": "Constantes mises à jour", "result": self.mathematical_constants}
+        elif message.get("type") == "update_resources":
+            self.resources.update(message.get("resources", {}))
+            return {"status": "Ressources mises à jour", "result": self.resources}
+        elif message.get("type") == "update_external_apis":
+            self.external_apis.update(message.get("apis", {}))
+            return {"status": "APIs mises à jour", "result": self.external_apis}

satellites/shakaSat/shaka.py

@@ -1,5 +1,5 @@
 from satellites.base_satellite import VegapunkSatellite
-from typing import Dict, Any, List
+from typing import Dict, Any, List, Optional
 import nltk
 from nltk.tokenize import word_tokenize
 from nltk.corpus import stopwords
@@ -163,13 +163,6 @@ class Shaka(VegapunkSatellite):
     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"}
-
-    def update_from_punkrecord(self):
-        print(f"{self.name} met à jour sa base de connaissances depuis Punkrecord")
-        self.add_to_knowledge_base("Last_update", "Nouveaux patterns détectés dans les données")
 
     def perform_advanced_analyse(self, data: Dict[str, Any]) -> Dict[str, Any]:
         text_analyse = self._analyse_text(data.get('text', ''))
@@ -180,3 +173,33 @@ class Shaka(VegapunkSatellite):
             "Analyse_logique": logical_analyse,
             "Conclusions": "Analyse intégrale basée sur le texte et les raisonnements logiques"
         }
+
+    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"}
+
+    def update_from_punkrecord(self):
+        print(f"{self.name} met à jour sa base de connaissances depuis Punkrecord")
+        self.add_to_knowledge_base("Last_update", "Nouveaux patterns détectés dans les données")
+        return {"Statut": "Mise à jour effectuée"}
+
+    def proccess_communicatioon(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        # logique a  shaka pour traiter le message
+        if message.get("type") == "ethics_check":
+            result = self.check_ethics(message.get("content"))
+            return {"status": "Traitement effectué", "result": result}
+        elif message.get("type") == "fact_check":
+            result = self.verify_facts(message.get("claim"))
+            return {"status": "Traitement effectué", "result": result}
+        elif message.get("type") == "task":
+            task_result = self.process_task(message.get("task"))
+            return {"status": "Traitement effectué", "result": task_result}
+        elif message.get("type") == "recommendations":
+            result = self.provide_ethical_recommendations(message.get("content"))
+            return {"status": "Traitement effectué", "result": result}
+        elif message.get("type") == "toggle_ethical_filter":
+            result = self.toggle_ethical_filter()
+            return {"status": "Traitement effectué", "result": result}
+
+
+

stellar/base_stellar.py

@@ -0,0 +1,32 @@
+from satellites.base_satellite import VegapunkSatellite
+from abc import ABC, abstractmethod
+from typing import  Dict, Any,Optional
+import logging
+
+
+
+
+class Stellar:
+    def __init__(self):
+        self.satellites = {} # dictionnaire pour stocke les references satellites
+
+
+    def register_satellites(self,satellite:VegapunkSatellite)->None:
+        self.satellites[satellite.name] = satellite
+
+    def route_communication(self, sender_name: str, target_name: str, message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        if sender_name not in self.satellites or target_name not in self.satellites:
+            logging.error(f"Impossible de router le message de {sender_name} à {target_name}")
+            return None
+
+        sender_name = self.satellites[sender_name]
+        target_name = self.satellites[target_name]
+
+        return sender_name.communicate_with_stellar(target_name, message)
+
+    def broadcast_message(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
+        results = {}
+        for satellite_name, satellite in self.satellites.items():
+            if satellite_name != sender_name:
+                results[satellite_name] = satellite.communicate_with_stellar(self.satellites[sender_name], message)
+        return results

tests/pythagoras_test.py

@@ -57,6 +57,28 @@ def test_pythagoras():
     stat_result = pythagoras.process_task(stat_task)
     print("Résultat du test statistique:", stat_result)
     print("\n\n")
+    # Exemple de données pour l'analyse
+    sample_data = [
+        {"x": 1, "y": 2, "z": 3},
+        {"x": 2, "y": 4, "z": 6},
+        {"x": 3, "y": 6, "z": 9},
+        {"x": 4, "y": 8, "z": 12},
+        {"x": 5, "y": 10, "z": 15}
+    ]
+
+    # Ajout de tâches à la file d'attente
+    pythagoras.add_task({"type": "analyze_data", "data": sample_data})
+    pythagoras.add_task({"type": "conduct_research", "topic": "Intelligence Artificielle", "depth": "deep"})
+    pythagoras.add_task({"type": "extract_information",
+                         "content": "L'intelligence artificielle (IA) est un domaine de l'informatique qui vise à créer des machines capables de simuler l'intelligence humaine. Elle englobe des sous-domaines tels que l'apprentissage automatique, le traitement du langage naturel et la vision par ordinateur.",
+                         "keywords": ["apprentissage automatique", "traitement du langage naturel"]})
+
+    # Traitement des tâches
+    while task := pythagoras.get_next_task():
+        result = pythagoras.process_task(task)
+        print(f"Résultat de la tâche : {result}")
+
+    print("\n\n")
 
     print("Connexion au Punkrecord")
     pythagoras.update_from_punkrecord()