Merge pull request #375 from capocchi/version-5.1

Version 5.1

Author: capocchi

devsimpy/DEVSKernel/KAFKADEVS/InMemoryKafkaWorker.py

@@ -7,6 +7,8 @@
 import logging
 import time
 
+from abc import ABC, abstractmethod
+
 from DEVSKernel.KafkaDEVS.logconfig import LOGGING_LEVEL, worker_kafka_logger
 
 logger = logging.getLogger("DEVSKernel.KafkaDEVS.InMemoryKafkaWorker")
@@ -21,24 +23,22 @@
 class InMemoryKafkaWorker(threading.Thread):
 	"""Worker thread that manages one atomic model in memory."""
 
-	def __init__(self, aDEVS, index, bootstrap_servers, in_topic=None, out_topic=None):
+	def __init__(self, model_name, aDEVS, bootstrap_server, in_topic=None, out_topic=None):
 		super().__init__(daemon=True)
 		self.aDEVS = aDEVS
-		self.index = index
-		self.bootstrap_servers = bootstrap_servers
+		self.model_name = model_name
+		self.bootstrap_server = bootstrap_server
 		self.running = True
 
 		# Topics explicitement fournis par la stratégie
 		self.in_topic = in_topic  ### from coodinator
 		self.out_topic = out_topic ### to coordinator
-	
-		self.aBlock = aDEVS.getBlockModel()
 
-		group_id = f"worker-thread-{index}-{int(time.time() * 1000)}"
+		group_id = f"worker-thread-{self.aDEVS.myID}-{int(time.time() * 1000)}"
 
 		# Kafka consumer pour le topic de travail dédié
 		self.consumer = Consumer({
-			"bootstrap.servers": bootstrap_servers,
+			"bootstrap.servers": bootstrap_server,
 			"group.id": group_id,
 			"auto.offset.reset": "latest",
 			"enable.auto.commit": True,
@@ -47,24 +47,20 @@ def __init__(self, aDEVS, index, bootstrap_servers, in_topic=None, out_topic=Non
 
 		# Kafka producer pour renvoyer les réponses
 		self.producer = Producer({
-			"bootstrap.servers": bootstrap_servers
+			"bootstrap.servers": bootstrap_server
 		})
 
 		logger.info(
             "  [Thread-%s] Created for model %s (in topic=%s, out topic=%s)",
-            index, self.aBlock.label, self.in_topic, self.out_topic
+            self.aDEVS.myID, self.model_name, self.in_topic, self.out_topic
         )
-
-	def get_index(self):
-		"""Returns the index of the model"""
-		return self.index
 
 	def get_model(self):
 		"""Returns the atomic DEVS model managed by this worker."""
 		return self.aDEVS
 
 	def get_model_label(self):
-		return self.aDEVS.getBlockModel().label
+		return self.model_name
 
 	def get_model_time_next(self):
 		return self.aDEVS.timeNext
@@ -132,12 +128,23 @@ def do_output_function(self):
 
 		self.aDEVS.outputFnc()
 
+	@abstractmethod
+	def _process_standard(self, data):
+		"""
+		Process standard DEVS message format.
+		Must be implemented by subclasses.
+		
+		Args:
+			data (dict): Parsed JSON message data
+		"""
+		pass
+
 	# ------------------------------------------------------------------
 	#  Boucle principale
 	# ------------------------------------------------------------------
 
 	def run(self):
-		logger.info("  [Thread-%s] Started", self.index)
+		logger.info(f"  [Thread-{self.aDEVS.myID}] Started")
 
 		while self.running:
 			msg = self.consumer.poll(timeout=0.5)
@@ -148,20 +155,15 @@ def run(self):
 				raw = msg.value().decode("utf-8")
 				data = json.loads(raw)
 
-				worker_kafka_logger.debug(
-					"[Thread-%s] IN: topic=%s value=%s",
-					self.index,
-					msg.topic(),
-					raw,
-				)
+				worker_kafka_logger.debug(f"[Thread-{self.aDEVS.myID}] IN: topic={msg.topic()} value={raw}")
 
 				self._process_standard(data)
 
 			except Exception as e:
-				logger.exception("[Thread-%s] Error in run loop: %s", self.index, e)
+				logger.exception("[Thread-%s] Error in run loop: %s", self.aDEVS.myID, e)
 
 		self.consumer.close()
-		logger.info("  [Thread-%s] Stopped", self.index)
+		logger.info(f"  [Thread-{self.aDEVS.myID}] Stopped")
 
 
 	def stop(self):

devsimpy/DEVSKernel/KAFKADEVS/MS4Me/DockerFile

@@ -0,0 +1,14 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends build-essential && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["python", "--version"]  # Commande par défaut

devsimpy/DEVSKernel/KAFKADEVS/MS4Me/MS4MeKafkaWorker.py

@@ -30,33 +30,25 @@ class MS4MeKafkaWorker(InMemoryKafkaWorker):
 
 	OUT_TOPIC = "ms4meOut"
 
-	def __init__(self, aDEVS, index, bootstrap_servers):
-		super().__init__(aDEVS, index, bootstrap_servers, in_topic=f"ms4me{aDEVS.getBlockModel().label}In", out_topic=MS4MeKafkaWorker.OUT_TOPIC)
+	def __init__(self, model_name, aDEVS, bootstrap_server):
+		""" Constructor
+		"""
+		super().__init__(model_name, aDEVS, bootstrap_server, in_topic=f"ms4me{model_name}In", out_topic=MS4MeKafkaWorker.OUT_TOPIC)
 
 		self.wire = StandardWireAdapter
 
 	def get_topic_to_write(self) -> str:
+		""" Return the topic used to contact the model
 		"""
-		Retourne le nom de topic à utiliser pour contacter ce modèle (obj)
-		"""
-		
 		return self.in_topic
-	
-	@staticmethod
-	def get_topic_to_read() -> str:
-		"""
-		Retourne le nom de topic à utiliser pour lire les résultats de tous les modèles
+
+	def get_topic_to_read(self) -> str:
+		""" Return the name of the topic to used to read the results od the model
 		"""
-		return MS4MeKafkaWorker.OUT_TOPIC
+		return self.out_topic
 
 	def output_msg_mapping(self) -> ModelOutputMessage:
-		"""
-
-		Args:
-			aDEVS (AtomicDEVS): DEVS atomic model from DomainBehaviorInterface
-
-		Returns:
-			ModelOutputMessage: ModelOutputMessage containing the output port values from Ms4me
+		""" Returns aModelOutputMessage message that contain the output port values from Ms4me
 		"""
 
 		result_portvalue_list = []
@@ -85,16 +77,15 @@ def output_msg_mapping(self) -> ModelOutputMessage:
 		return result_portvalue_list
 
 	def _handle_devs_message(self, msg: BaseMessage) -> BaseMessage:
-		"""
-		Reçoit un BaseMessage (InitSim, ExecuteTransition, SendOutput, ...)
-		et renvoie un BaseMessage de réponse (NextTime, ModelOutputMessage, ...).
+		""" Receive a BaseMessage (InitSim, ExecuteTransition, SendOutput, ...)
+		and send a BasMessage in repsonce (NextTime, ModelOutputMessage, ...).
 		"""
 		t = msg.time.t
 
 		# --- InitSim : initialisation + timeAdvance initial ---
 		if isinstance(msg, InitSim):
 			self.do_initialize(t)
-			return NextTime(SimTime(t=float(self.aDEVS.timeNext)), sender=self.aBlock.label)
+			return NextTime(SimTime(t=float(self.aDEVS.timeNext)), sender=self.model_name)
 
 		# --- ExecuteTransition
 		if isinstance(msg, ExecuteTransition):
@@ -146,7 +137,7 @@ def temp_peek(port, *args):
 
 			ta = float(ta) if ta is not None else float("inf")
 
-			return TransitionDone(time=SimTime(t=t), nextTime=SimTime(t=ta), sender=self.aBlock.label)
+			return TransitionDone(time=SimTime(t=t), nextTime=SimTime(t=ta), sender=self.model_name)
 
 		# --- SendOutput : outputFnc + calcul des sorties ---
 		if isinstance(msg, SendOutput):
@@ -159,42 +150,39 @@ def temp_peek(port, *args):
 			return ModelOutputMessage(
 				modelOutput = port_values,
 				nextTime = next_time,
-				sender = self.aBlock.label,
+				sender = self.model_name,
 			)		
 
 		if isinstance(msg, SimulationDone):
 			self.running = False
 			return ModelDone(
 				time = msg.time,
-				sender = self.aDEVS.getBlockModel().label,
+				sender = self.model_name,
 			)
-
-		# --- NextTime, ModelOutputMessage, etc. ---
-		# Ces types sont normalement utilisés comme réponses, pas comme requêtes
-		# côté worker, donc on ne les traite pas ici.
+		
 		raise ValueError(f"Unsupported message type in worker: {type(msg).__name__}")
 
 	def _process_standard(self, data):
 		"""Process standard DEVS message format."""
 
 		devs_msg = self.wire.from_wire(data)
 
-		# Traiter le message DEVS
+		# handel the DEVS msessage
 		try:
 			reply_msg = self._handle_devs_message(devs_msg)
 		except Exception as e:
-			logger.exception("  [Thread-%s] Error handling message: %s", self.index, e)
+			logger.exception("  [Thread-%s] Error handling message: %s", self.aDEVS.myID, e)
 
-		# Préparer le message de réponse
+		# define a msg to send
 		reply_wire = self.wire.to_wire(reply_msg)					
 		reply_json = json.dumps(reply_wire).encode("utf-8")
 
-		# Log et envoi de la réponse
+		# Log
 		worker_kafka_logger.debug(
-			f"[Thread-{self.index}] OUT: topic={self.out_topic} value={reply_json.decode('utf-8')}"
+			f"[Thread-{self.aDEVS.myID}] OUT: topic={self.out_topic} value={reply_json.decode('utf-8')}"
 		)
 
-		# Envoi de la réponse
+		# send the msg to the out_topic
 		self.producer.produce(
 			self.out_topic,
 			value=reply_json,

devsimpy/DEVSKernel/KAFKADEVS/MS4Me/ms4me_kafka_messages.py

@@ -3,25 +3,30 @@
 from __future__ import annotations
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
-
+import numpy as np
 # ---------- Types de base ----------
 
 @dataclass
 class SimTime:
-    timeType: str = "devs.msg.time.LongSimTime"
+    timeType: str = "devs.msg.time.DoubleSimTime"
     t: float = 0.0
 
     def to_dict(self) -> Dict[str, Any]:
+        if self.t == float("inf"): 
+            self.t = np.finfo(np.float64).max
         return {
             "timeType": self.timeType,
             "t": self.t,
         }
 
     @staticmethod
     def from_dict(d: Dict[str, Any]) -> "SimTime":
+        time =float(d.get("t", 0.0))
+        if time == np.finfo(np.float64).max:
+            time = float("inf")
         return SimTime(
-            timeType=d.get("timeType", "devs.msg.time.LongSimTime"),
-            t=float(d.get("t", 0.0)),
+            timeType=d.get("timeType", "devs.msg.time.DoubleSimTime"),
+            t=time,
         )