1+ import logging
2+ import json
3+
4+ from DEVSKernel .KafkaDEVS .logconfig import LOGGING_LEVEL , worker_kafka_logger
5+
6+ logger = logging .getLogger ("DEVSKernel.KafkaDEVS.InMemoryKafkaWorker" )
7+ logger .setLevel (LOGGING_LEVEL )
8+
9+ from DEVSKernel .KafkaDEVS .InMemoryKafkaWorker import InMemoryKafkaWorker
10+
11+ from .ms4me_kafka_messages import (
12+ BaseMessage ,
13+ SimTime ,
14+ InitSim ,
15+ ExecuteTransition ,
16+ SendOutput ,
17+ NextTime ,
18+ TransitionDone ,
19+ ModelOutputMessage ,
20+ PortValue ,
21+ SimulationDone ,
22+ ModelDone ,
23+ )
24+
25+ from .ms4me_kafka_wire_adapters import StandardWireAdapter
26+ from DomainInterface .Object import Message
27+
28+ class MS4MeKafkaWorker (InMemoryKafkaWorker ):
29+ """Worker thread that manages one atomic model in memory."""
30+
31+ OUT_TOPIC = "ms4meOut"
32+
33+ def __init__ (self , aDEVS , index , bootstrap_servers ):
34+ super ().__init__ (aDEVS , index , bootstrap_servers , in_topic = f"ms4me{ aDEVS .getBlockModel ().label } , out_topic = MS4MeKafkaWorker .OUT_TOPIC )
35+
36+ self .wire = StandardWireAdapter
37+
38+ def get_topic_to_write (self ) -> str :
39+ """
40+ Retourne le nom de topic à utiliser pour contacter ce modèle (obj)
41+ """
42+
43+ return self .in_topic
44+
45+ @staticmethod
46+ def get_topic_to_read () -> str :
47+ """
48+ Retourne le nom de topic à utiliser pour lire les résultats de tous les modèles
49+ """
50+ return MS4MeKafkaWorker .OUT_TOPIC
51+
52+ def output_msg_mapping (self ) -> ModelOutputMessage :
53+ """
54+
55+ Args:
56+ aDEVS (AtomicDEVS): DEVS atomic model from DomainBehaviorInterface
57+
58+ Returns:
59+ ModelOutputMessage: ModelOutputMessage containing the output port values from Ms4me
60+ """
61+
62+ result_portvalue_list = []
63+ for port , m in self .aDEVS .myOutput .items ():
64+ if isinstance (m , Message ):
65+ value = getattr (m , "value" , m )
66+ port_name = getattr (port , "name" , str (port ))
67+ result_portvalue_list .append (
68+ PortValue (
69+ value = value ,
70+ portIdentifier = port_name ,
71+ portType = type (value ).__name__ ,
72+ )
73+ )
74+ else :
75+ value = m
76+ port_name = getattr (port , "name" , str (port ))
77+ result_portvalue_list .append (
78+ PortValue (
79+ value = value ,
80+ portIdentifier = port_name ,
81+ portType = type (value ).__name__ ,
82+ )
83+ )
84+
85+ return result_portvalue_list
86+
87+ def _handle_devs_message (self , msg : BaseMessage ) -> BaseMessage :
88+ """
89+ Reçoit un BaseMessage (InitSim, ExecuteTransition, SendOutput, ...)
90+ et renvoie un BaseMessage de réponse (NextTime, ModelOutputMessage, ...).
91+ """
92+ t = msg .time .t
93+
94+ # --- InitSim : initialisation + timeAdvance initial ---
95+ if isinstance (msg , InitSim ):
96+ self .do_initialize (t )
97+ return NextTime (SimTime (t = float (self .aDEVS .timeNext )), sender = self .aBlock .label )
98+
99+ # --- ExecuteTransition
100+ if isinstance (msg , ExecuteTransition ):
101+ t = msg .time .t
102+
103+ # Si on a des inputs, c'est une extTransition
104+ if msg .portValueList :
105+ port_inputs = {}
106+
107+ # Construire dict {port_obj -> Message(value, time)}
108+ from DomainInterface .Object import Message
109+ for pv in msg .portValueList :
110+ # pv.portIdentifier doit matcher le nom du port d'entrée
111+ for iport in self .aDEVS .IPorts :
112+ if iport .name == pv .portIdentifier :
113+ m = Message (pv .value , t )
114+ port_inputs [iport ] = m
115+ break
116+
117+ # Sauvegarder l'ancien peek()
118+ old_peek = getattr (self .aDEVS , "peek" , None )
119+
120+ def temp_peek (port , * args ):
121+ if args and isinstance (args [0 ], dict ):
122+ return args [0 ].get (port )
123+ return port_inputs .get (port )
124+
125+ # Override temporairement peek()
126+ self .aDEVS .peek = temp_peek
127+
128+ try :
129+ # with self._temporary_peek(port_inputs):
130+ self .do_external_transition (t , msg )
131+ # self.aDEVS.extTransition(port_inputs)
132+ finally :
133+ if old_peek is not None :
134+ self .aDEVS .peek = old_peek
135+ else :
136+ # On enlève l'attribut si inexistant avant
137+ if hasattr (self .aDEVS , "peek" ):
138+ delattr (self .aDEVS , "peek" )
139+ else :
140+ # Pas d'inputs : transition interne
141+ # self.aDEVS.intTransition()
142+ self .do_internal_transition (t )
143+
144+ # Après extTransition, on recalcule le ta
145+ ta = self .aDEVS .timeNext
146+
147+ ta = float (ta ) if ta is not None else float ("inf" )
148+
149+ return TransitionDone (time = SimTime (t = t ), nextTime = SimTime (t = ta ), sender = self .aBlock .label )
150+
151+ # --- SendOutput : outputFnc + calcul des sorties ---
152+ if isinstance (msg , SendOutput ):
153+ t = msg .time .t
154+
155+ self .do_output_function ()
156+ port_values = self .output_msg_mapping ()
157+
158+ next_time = SimTime (t = t )
159+ return ModelOutputMessage (
160+ modelOutput = port_values ,
161+ nextTime = next_time ,
162+ sender = self .aBlock .label ,
163+ )
164+
165+ if isinstance (msg , SimulationDone ):
166+ self .running = False
167+ return ModelDone (
168+ time = msg .time ,
169+ sender = self .aDEVS .getBlockModel ().label ,
170+ )
171+
172+ # --- NextTime, ModelOutputMessage, etc. ---
173+ # Ces types sont normalement utilisés comme réponses, pas comme requêtes
174+ # côté worker, donc on ne les traite pas ici.
175+ raise ValueError (f"Unsupported message type in worker: { type (msg ).__name__ } )
176+
177+ def _process_standard (self , data ):
178+ """Process standard DEVS message format."""
179+
180+ devs_msg = self .wire .from_wire (data )
181+
182+ # Traiter le message DEVS
183+ try :
184+ reply_msg = self ._handle_devs_message (devs_msg )
185+ except Exception as e :
186+ logger .exception (" [Thread-%s] Error handling message: %s" , self .index , e )
187+
188+ # Préparer le message de réponse
189+ reply_wire = self .wire .to_wire (reply_msg )
190+ reply_json = json .dumps (reply_wire ).encode ("utf-8" )
191+
192+ # Log et envoi de la réponse
193+ worker_kafka_logger .debug (
194+ f"[Thread-{ self .index } { self .out_topic } { reply_json .decode ("utf-8" )}
195+ )
196+
197+ # Envoi de la réponse
198+ self .producer .produce (
199+ self .out_topic ,
200+ value = reply_json ,
201+ )
202+ self .producer .flush ()
0 commit comments