deploy: f609b3f

daq/jobs/handle_stats.html

@@ -86,13 +86,9 @@ <h2 class="section-title" id="header-classes">Classes</h2>
             return True
 
         if isinstance(message, DAQJobMessageStats):
-            self._stats[message.daq_job_info.supervisor_config.supervisor_id] = (
-                message.stats
-            )
+            self._stats[message.supervisor_id] = message.stats
         elif isinstance(message, DAQJobMessageStatsRemote):
-            self._remote_stats[message.daq_job_info.supervisor_config.supervisor_id] = (
-                message.stats
-            )
+            self._remote_stats[message.supervisor_id] = message.stats
         return True
 
     def _save_stats(self):
@@ -147,9 +143,9 @@ <h2 class="section-title" id="header-classes">Classes</h2>
             &#34;is_alive&#34;,
             &#34;last_active&#34;,
             &#34;message_in_count&#34;,
-            &#34;message_in_bytes&#34;,
+            &#34;message_in_megabytes&#34;,
             &#34;message_out_count&#34;,
-            &#34;message_out_bytes&#34;,
+            &#34;message_out_megabytes&#34;,
         ]
         data_to_send = []
 
@@ -164,12 +160,19 @@ <h2 class="section-title" id="header-classes">Classes</h2>
             ].items():
                 remote_stats_combined[supervisor_id] = remote_stats
 
-        for remote_stats_dict in self._remote_stats.values():
+        for remote_supervisor_id, remote_stats_dict in self._remote_stats.items():
             # For each remote stats dict, combine the values
             for (
                 supervisor_id,
                 remote_stats_dict_serialized_item,
             ) in remote_stats_dict.items():
+                # Skip if the remote supervisor id is the same as the local supervisor id or
+                # other supervisors try to overwrite other supervisors
+                if supervisor_id != remote_supervisor_id or (
+                    self._supervisor_config
+                    and self._supervisor_config.supervisor_id == remote_supervisor_id
+                ):
+                    continue
                 # Convert the supervisor remote stats to a dict
                 remote_stats_dict_serialized = msgspec.structs.asdict(
                     remote_stats_dict_serialized_item
@@ -179,20 +182,23 @@ <h2 class="section-title" id="header-classes">Classes</h2>
                     if value == 0 or not value:
                         continue
                     setattr(remote_stats_combined[supervisor_id], item, value)
-
         for supervisor_id, remote_stats in remote_stats_combined.items():
             is_remote_alive = datetime.now() - remote_stats.last_active &lt;= timedelta(
                 seconds=DAQ_JOB_HANDLE_STATS_REMOTE_ALIVE_SECONDS
             )
+
+            def _byte_to_mb(x):
+                return &#34;{:.3f}&#34;.format(x / 1024 / 1024)
+
             data_to_send.append(
                 [
                     supervisor_id,
                     str(is_remote_alive).lower(),
                     remote_stats.last_active,
                     remote_stats.message_in_count,
-                    remote_stats.message_in_bytes,
+                    _byte_to_mb(remote_stats.message_in_bytes),
                     remote_stats.message_out_count,
-                    remote_stats.message_out_bytes,
+                    _byte_to_mb(remote_stats.message_out_bytes),
                 ]
             )
         self._put_message_out(

daq/jobs/remote.html

@@ -327,8 +327,8 @@ <h2 id="attributes">Attributes</h2>
 
     def _send_remote_stats_message(self):
         msg = DAQJobMessageStatsRemote(dict(self._remote_stats))
-        self._send_remote_message(msg)
         self._put_message_out(msg)
+        self.message_in.put(msg)
 
     def __del__(self):
         &#34;&#34;&#34;

daq/models.html

@@ -227,7 +227,14 @@ <h2 id="attributes">Attributes</h2>
     timestamp: Optional[datetime] = field(default_factory=datetime.now)
     is_remote: bool = False
     daq_job_info: Optional[&#34;DAQJobInfo&#34;] = None
-    remote_config: DAQRemoteConfig = field(default_factory=DAQRemoteConfig)</code></pre>
+    remote_config: DAQRemoteConfig = field(default_factory=DAQRemoteConfig)
+
+    @property
+    def supervisor_id(self) -&gt; str:
+        if self.daq_job_info is None or self.daq_job_info.supervisor_config is None:
+            return &#34;unknown&#34;
+
+        return self.daq_job_info.supervisor_config.supervisor_id</code></pre>
 </details>
 <h3>Ancestors</h3>
 <ul class="hlist">
@@ -260,6 +267,21 @@ <h3>Instance variables</h3>
 <dd>
 <div class="desc"></div>
 </dd>
+<dt id="enrgdaq.daq.models.DAQJobMessage.supervisor_id"><code class="name">prop <span class="ident">supervisor_id</span> : str</code></dt>
+<dd>
+<div class="desc"></div>
+<details class="source">
+<summary>
+<span>Expand source code</span>
+</summary>
+<pre><code class="python">@property
+def supervisor_id(self) -&gt; str:
+    if self.daq_job_info is None or self.daq_job_info.supervisor_config is None:
+        return &#34;unknown&#34;
+
+    return self.daq_job_info.supervisor_config.supervisor_id</code></pre>
+</details>
+</dd>
 <dt id="enrgdaq.daq.models.DAQJobMessage.timestamp"><code class="name">var <span class="ident">timestamp</span> : datetime.datetime | None</code></dt>
 <dd>
 <div class="desc"></div>
@@ -600,11 +622,12 @@ <h4><code><a title="enrgdaq.daq.models.DAQJobInfo" href="#enrgdaq.daq.models.DAQ
 </li>
 <li>
 <h4><code><a title="enrgdaq.daq.models.DAQJobMessage" href="#enrgdaq.daq.models.DAQJobMessage">DAQJobMessage</a></code></h4>
-<ul class="">
+<ul class="two-column">
 <li><code><a title="enrgdaq.daq.models.DAQJobMessage.daq_job_info" href="#enrgdaq.daq.models.DAQJobMessage.daq_job_info">daq_job_info</a></code></li>
 <li><code><a title="enrgdaq.daq.models.DAQJobMessage.id" href="#enrgdaq.daq.models.DAQJobMessage.id">id</a></code></li>
 <li><code><a title="enrgdaq.daq.models.DAQJobMessage.is_remote" href="#enrgdaq.daq.models.DAQJobMessage.is_remote">is_remote</a></code></li>
 <li><code><a title="enrgdaq.daq.models.DAQJobMessage.remote_config" href="#enrgdaq.daq.models.DAQJobMessage.remote_config">remote_config</a></code></li>
+<li><code><a title="enrgdaq.daq.models.DAQJobMessage.supervisor_id" href="#enrgdaq.daq.models.DAQJobMessage.supervisor_id">supervisor_id</a></code></li>
 <li><code><a title="enrgdaq.daq.models.DAQJobMessage.timestamp" href="#enrgdaq.daq.models.DAQJobMessage.timestamp">timestamp</a></code></li>
 </ul>
 </li>