ajoute et implementation de la communication etre Satellite

main.py

@@ -24,9 +24,11 @@ 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"})
 
+# Exemple de communication
+response = stellar.route_communication("Shaka", "Atlas", {"type": "monitor_directory", "content": "Suspicious activity detected"})
+print(response)
 
+# Exemple de diffusion
+results = stellar.broadcast_message("Shaka", {"type": "ethics_update", "content": "New ethical guidelines"})
 print(results)

satellites/atlasSat/atlas.py

@@ -127,7 +127,7 @@ class Atlas(VegapunkSatellite):
 
 
 
-    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
+    def process_communication(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
 
         if message.get("type") == "task":
             task_result = self.process_task(message.get("task"))
@@ -152,4 +152,10 @@ class Atlas(VegapunkSatellite):
             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}
+            return {"status": "Changements vérifiés", "result": changes}
+        else:
+            return {"status": "Erreur", "result": "Type de tâche inconnu"}
+
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)

satellites/base_satellite.py

@@ -74,13 +74,12 @@ class VegapunkSatellite(ABC):
             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}
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} a recu un message de : {sender_name}")
+        return self.process_communication(sender_name, message)
 
     @abstractmethod
-    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any])->Dict[str,Any]:
+    def process_communication(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

satellites/edisonSat/edison.py

@@ -123,3 +123,26 @@ class Edison(VegapunkSatellite):
     def update_from_punkrecord(self) -> None:
         self.log_activity("Mise à jour depuis PunkRecord")
         # Ici, vous pourriez implémenter la logique pour mettre à jour les API externes ou les paramètres du modèle LLM
+
+
+    def process_communication(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") == "innovation":
+            innovation_result = self.generate_innovation(message.get("domain"))
+            return {"status": "Traitement effectué", "result": innovation_result}
+        elif message.get("type") == "logic_problem":
+            logic_result = self.solve_logic_problem(message.get("problem"))
+            return {"status": "Traitement effectué", "result": logic_result}
+        elif message.get("type") == "data_analysis":
+            analysis_result = self.analyze_data(message.get("data"))
+            return {"status": "Traitement effectué", "result": analysis_result}
+        else:
+            return {"status": "Erreur", "result": "Type de tâche inconnu"}
+
+
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)
+

satellites/lilithSat/lilith.py

@@ -14,10 +14,12 @@ class Lilith(VegapunkSatellite):
         super().__init__(name="Lilith", specialty=role)
         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"]
+        self.unconventional_approaches = ["Pensée inversée", "Analogies lointaines", "Combinaison aléatoire",
+                                          "Contraintes extrêmes"]
         # 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":
@@ -114,8 +116,7 @@ class Lilith(VegapunkSatellite):
         self.log_activity("Mise à jour depuis PunkRecord")
         # 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]:
+    def process_communication(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}
@@ -134,4 +135,9 @@ class Lilith(VegapunkSatellite):
         elif message.get("type") == "status_report":
             result = self.report_status()
             return {"status": "Rapport de status généré", "result": result}
+        else:
+            return {"status": "Erreur", "result": "Type de tâche inconnu"}
 
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)

satellites/pythagorasSat/pythagoras.py

@@ -20,11 +20,11 @@ load_dotenv()
 
 
 # client = OpenAI(os.getenv("LLM_API_KEY"))
-
-client =OpenAI(
-    organization="LLM_API_KEY",
-)
-
+#
+# client =OpenAI(
+#     organization="LLM_API_KEY",
+# )
+#
 
 role = "mathématiques, statistiques et analyse de données"
 fonction = "Effectuer des calculs complexes et analyser des ensembles de données"
@@ -248,7 +248,7 @@ class Pythagoras(VegapunkSatellite):
         return status
 
 
-    def proccess_communicatioon(self,sender_name:str,message:Dict[str,Any]) ->Dict[str,Any]:
+    def process_communication(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}
@@ -267,3 +267,11 @@ class Pythagoras(VegapunkSatellite):
         elif message.get("type") == "update_external_apis":
             self.external_apis.update(message.get("apis", {}))
             return {"status": "APIs mises à jour", "result": self.external_apis}
+
+        else:
+            return {"status": "Erreur", "result": "Type de tâche inconnu"}
+
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)
+

satellites/shakaSat/shaka.py

@@ -183,23 +183,33 @@ class Shaka(VegapunkSatellite):
         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
+    def process_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        # Logique spécifique à Shaka pour traiter les communications
         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}
+            # Effectuer une vérification éthique
+            result = self.perform_ethics_check(message.get("content"))
+            return {"status": "completed", "result": result}
+        elif message.get("type") == "security_alert":
+            # Traiter une alerte de sécurité du point de vue éthique
+            analysis = self.analyze_security_alert(message.get("content"))
+            return {"status": "analyzed", "ethical_analysis": analysis}
+        else:
+            return {"status": "unknown_message_type"}
+
+
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)
+
+    def perform_ethics_check(self, content: str) -> str:
+        # Implémentez ici la logique de vérification éthique
+        # Ceci est un exemple simplifié
+        return f"Ethical analysis of: {content}"
+
+    def analyze_security_alert(self, alert_content: str) -> str:
+        # Implémentez ici la logique d'analyse éthique des alertes de sécurité
+        # Ceci est un exemple simplifié
+        return f"Ethical implications of security alert: {alert_content}"
 
 
 

satellites/yorkSat/york.py

@@ -111,3 +111,28 @@ class York(VegapunkSatellite):
     def update_from_punkrecord(self) -> None:
         self.log_activity("Mise à jour depuis PunkRecord")
         # Ici, vous pourriez implémenter la logique pour mettre à jour les seuils de ressources ou les plannings de maintenance
+
+    def process_communication(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") == "resource_thresholds":
+            self.resource_thresholds = message.get("thresholds")
+            return {"status": "Seuils de ressources mis à jour", "result": self.resource_thresholds}
+        elif message.get("type") == "maintenance_schedule":
+            self.maintenance_schedule = message.get("schedule")
+            return {"status": "Planning de maintenance mis à jour", "result": self.maintenance_schedule}
+        elif message.get("type") == "external_api":
+            api_name = message.get("api_name")
+            api_config = message.get("api_config")
+            self.external_apis[api_name] = api_config
+            return {"status": "API externe ajoutée", "result": api_config}
+        else:
+            return {"status": "Message non reconnu", "result": "Aucune action effectuée"}
+
+
+
+    def receive_communication(self, sender_name: str, message: Dict[str, Any]) -> Dict[str, Any]:
+        logging.info(f"{self.name} received communication from {sender_name}")
+        return self.process_communication(sender_name, message)
+

stellar/base_stellar.py

@@ -16,17 +16,17 @@ class Stellar:
 
     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}")
+            logging.error(f"Invalid sender or target: {sender_name} -> {target_name}")
             return None
 
-        sender_name = self.satellites[sender_name]
-        target_name = self.satellites[target_name]
+        sender = self.satellites[sender_name]
+        target = self.satellites[target_name]
 
-        return sender_name.communicate_with_stellar(target_name, message)
+        return sender.communicate_with_other_satellite(target, 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)
+                results[satellite_name] = self.route_communication(sender_name, satellite_name, message)
         return results