Basic diagnostics

Author: Tom-Willemsen

rust/src/lib.rs

@@ -28,7 +28,7 @@ fn tof_bin_event_linear(tof: i32, start: i32, stop: i32, step: i32) -> Option<us
 
 #[pyfunction]
 fn bin_events_into_spectrum(
-    mut histogram: PyReadwriteArray2<u64>,
+    mut histogram: PyReadwriteArray2<f64>,
     event_tofs: PyReadonlyArray1<i32>,
     pixel_ids: PyReadonlyArray1<i32>,
     tof_bin_boundaries: PyReadonlyArray1<i32>,
@@ -42,7 +42,7 @@ fn bin_events_into_spectrum(
         .for_each(|(&tof, &pixel)| {
             if let Some(tof_bin) = tof_bin_event(tof, boundaries)
                 && let Some(e) = histogram.get_mut((pixel as usize, tof_bin)) {
-                *e += 1;
+                *e += 1.0
             }
         });
 
@@ -51,7 +51,7 @@ fn bin_events_into_spectrum(
 
 #[pyfunction]
 fn bin_events_into_spectrum_linear(
-    mut histogram: PyReadwriteArray2<u64>,
+    mut histogram: PyReadwriteArray2<f64>,
     event_tofs: PyReadonlyArray1<i32>,
     pixel_ids: PyReadonlyArray1<i32>,
     tof_bin_start: i32,
@@ -66,7 +66,7 @@ fn bin_events_into_spectrum_linear(
         .for_each(|(&tof, &pixel)| {
             if let Some(tof_bin) = tof_bin_event_linear(tof, tof_bin_start, tof_bin_stop, tof_bin_step)
                 && let Some(e) = histogram.get_mut((pixel as usize, tof_bin)) {
-                *e += 1;
+                *e += 1.0
             }
         });
 

src/kafka_dae_diagnostics/_cli.py

@@ -6,13 +6,16 @@
 
 import numpy as np
 from confluent_kafka.cimpl import TopicPartition
-from p4p.server import DynamicProvider, Server
+from p4p.server import DynamicProvider, Server, StaticProvider
 from confluent_kafka import Consumer
 
 from kafka_dae_diagnostics.data import Data
 from kafka_dae_diagnostics.kafka_handlers import handle_event_messages, \
-    handle_run_info_messages, handle_event_msg
+    handle_run_info_messages
 from kafka_dae_diagnostics.spectrum_handlers import SpectrumHandler
+import p4p
+
+from kafka_dae_diagnostics.static_pvs import StaticPVs
 
 logger = logging.getLogger(__name__)
 
@@ -21,31 +24,51 @@
 
 def main() -> None:
     data = Data(
-        spectra=np.zeros(shape=(1, 1, 1), dtype=np.uint64),
-        spectrum_updaters=[],
+        spectra=np.zeros(shape=(1, 1, 1), dtype=np.float64),
+        callbacks={},
+        bin_boundaries=np.linspace(0, 20_000_000, num=1001, dtype=np.int32)
     )
 
-    handler = SpectrumHandler("TE:NDW2922:KDAEDIAG:", data)
+    prefix = "TE:NDW2922:KDAEDIAG:"
+
+    static_pvs = StaticPVs(data)
+    static_provider = StaticProvider()
+    static_provider.add(f"{prefix}EVENTS", static_pvs.total_events)
+    static_provider.add(f"{prefix}MEVENTS", static_pvs.total_mevents)
+    static_provider.add(f"{prefix}TOTALCOUNTS", static_pvs.total_events)
+    static_provider.add(f"{prefix}TOTAL_EVENT_MESSAGES", static_pvs.total_event_messages)
+    static_provider.add(f"{prefix}TOTAL_EVENT_MEGABYTES", static_pvs.total_event_megabytes)
+    static_provider.add(f"{prefix}COUNTRATE", static_pvs.count_rate)
+    static_provider.add(f"{prefix}HISTMEMORY", static_pvs.histogram_memory)
+
+    static_provider.add(f"{prefix}NUMPERIODS", static_pvs.num_periods)
+    static_provider.add(f"{prefix}NUMSPECTRA", static_pvs.num_spectra)
+    static_provider.add(f"{prefix}NUMTIMECHANNELS", static_pvs.num_time_channels)
+
+    static_provider.add(f"{prefix}STARTTIME", static_pvs.start_time)
+    static_provider.add(f"{prefix}RUNDURATION", static_pvs.run_duration)
+    static_provider.add(f"{prefix}DIAGNOSTICSLAG", static_pvs.processing_lag)
+
+    spectrum_handler = SpectrumHandler(prefix, data)
     providers = [
-        DynamicProvider("spectra", handler=handler),
+        DynamicProvider("spectra", handler=spectrum_handler),
+        static_provider,
     ]
+
+    data.callbacks["static-callbacks"] = static_pvs.update_all
+
     server = Server(providers=providers)
     with server:
         consume_from_kafka_forever(data)
 
 
-def update_spectra(data: Data) -> None:
-    num_updaters = len(data.spectrum_updaters)
-    if num_updaters == 0:
-        return
-
-    logger.debug("Updating %d spectra for connected clients", num_updaters)
-    for period, detector, pv in data.spectrum_updaters:
-        try:
-            pv.post(data.spectra[(0, detector)].astype(np.double), timestamp=time.time())
-        except Exception as e:
-            logger.warning("Failed to update dynamic spectrum PV for period %d, detector %d, error: %s %s", period,
-                           detector, e.__class__.__name__, e)
+def callbacks(data: Data) -> None:
+    with data.callbacks_lock:
+        for callback_id, cb in data.callbacks.items():
+            try:
+                cb(data)
+            except Exception as e:
+                logger.warning("Callback '%s' failed, error: %s %s", callback_id, e.__class__.__name__, e)
 
 
 def make_runinfo_consumer(settings: dict[str, Any]) -> Consumer:
@@ -59,10 +82,8 @@ def make_runinfo_consumer(settings: dict[str, Any]) -> Consumer:
     """
     runinfo_consumer = Consumer(settings)
 
-    start_offset = (
-        runinfo_consumer.get_watermark_offsets(TopicPartition("NDW2922_runInfo", 0), cached=False)[1]
-        - 2
-    )
+    low, high = runinfo_consumer.get_watermark_offsets(TopicPartition("NDW2922_runInfo", 0), cached=False)
+    start_offset = max(high - 2, low)
     runinfo_consumer.assign([TopicPartition("NDW2922_runInfo", 0, start_offset)])
     return runinfo_consumer
 
@@ -92,12 +113,15 @@ def consume_from_kafka_forever(data: Data) -> None:
 
     while True:
         run_info_messages = runinfo_consumer.consume(num_messages=50, timeout=0.)
-        handle_run_info_messages(run_info_messages, data=data, event_consumer=event_consumer)
+        if run_info_messages:
+            handle_run_info_messages(run_info_messages, data=data, event_consumer=event_consumer)
 
         event_messages = event_consumer.consume(num_messages=1000, timeout=0.1)
-        handle_event_messages(event_messages, data=data)
+        if event_messages:
+            start = time.time()
+            handle_event_messages(event_messages, data=data)
+            logger.debug("Handled %d event messages in %.3f ms", len(event_messages), ((time.time() - start) * 1000))
 
         if len(event_messages) > 0 or len(run_info_messages) > 0:
-            # If any messages arrived, spectra may have changed - update any PVs which
-            # are connected.
-            update_spectra(data)
+            # If any messages arrived, data may have changed - update all subscribed callbacks
+            callbacks(data)

src/kafka_dae_diagnostics/_kdaediag_rs.pyi

@@ -5,7 +5,7 @@ def bin_events_into_spectrum(
     event_tofs: npt.NDArray,
     pixel_ids: npt.NDArray,
     tof_bin_boundaries: npt.NDArray,
-) -> npt.NDArray:
+) -> None:
     pass
 
 
@@ -16,5 +16,5 @@ def bin_events_into_spectrum_linear(
     tof_bin_start: int,
     tof_bin_stop: int,
     tof_bin_step: int,
-) -> npt.NDArray:
+) -> None:
     pass

src/kafka_dae_diagnostics/data.py

@@ -1,6 +1,10 @@
 """Data being served by this IOC."""
 
 import dataclasses
+import threading
+import time
+from datetime import date
+from typing import Callable
 
 import numpy.typing as npt
 import numpy as np
@@ -12,17 +16,96 @@
 class Data:
     """A mutable object describing the data being served."""
 
-    spectra: npt.NDArray[np.uint64]
+    spectra: npt.NDArray[np.uint]
     """
     An array describing counts since last run start.
 
     Has shape ``(n_periods, n_detectors, n_timechannels)`` and
-    data-type :py:obj:`numpy.uint64`.
+    data-type :py:obj:`numpy.uint`.
     """
 
-    spectrum_updaters: list[tuple[int, int, SharedPV]]
+    callbacks: dict[str, Callable[["Data"], None]]
+    """
+    A list of callbacks to notify when data is updated.
     """
-    A list of callbacks to notify when a spectrum is updated.
 
-    Arguments are (period, detector, :py:obj:`~p4p.server.thread.SharedPV`).
+    bin_boundaries: npt.NDArray[np.int32]
+    """
+    Time-bin boundaries (ns).
+    """
+
+    callbacks_lock: threading.RLock = threading.RLock()
+    """
+    Lock-object, must be taken when spectrum_updaters is iterated/mutated
+    :meta private:
+    """
+
+    total_events: int = 0
+    """
+    Total number of neutron events in this run.
+    """
+
+    total_event_messages: int = 0
+    """
+    Total number of ev44 event messages in this run.
+    """
+
+    total_event_megabytes: float = 0.
+    """
+    Megabytes of event messages processed in this run.
+    """
+
+    largest_kafka_timestamp: float = 0.
+    """
+    Largest timestamp seen in an ev44 or pl72 message since
+    the beginning of this run. Seconds since epoch.
     """
+
+    most_recent_kafka_timestamp: float = 0.
+    """
+    Timestamp in the most recently-processed ev44 or pl72 message.
+    Seconds since epoch.
+    """
+
+    start_time: float = 0.
+    """
+    Timestamp of the most recent pl72 run-start message.
+    Seconds since epoch.
+    """
+
+    processing_lag: float = 0.
+    """
+    Estimated time difference between an event being recorded in
+    electronics and processed in KDAEDIAG ioc.
+    """
+
+    @property
+    def mev(self):
+        return self.total_events / 1_000_000
+
+    @property
+    def duration(self):
+        return max(self.largest_kafka_timestamp - self.start_time, 0)
+
+    @property
+    def mev_per_hour(self):
+        duration = self.duration
+        if duration == 0:
+            return 0
+        return (self.mev / duration) * 3600
+
+    @property
+    def num_periods(self) -> int:
+        return self.spectra.shape[0]
+
+    @property
+    def num_detectors(self) -> int:
+        return self.spectra.shape[1]
+
+    @property
+    def num_time_channels(self) -> int:
+        return self.spectra.shape[2]
+
+    @property
+    def histogram_megabytes(self) -> int:
+        return self.spectra.nbytes / 1024**2

src/kafka_dae_diagnostics/kafka_handlers.py

@@ -2,6 +2,9 @@
 Utilities for reacting to Kafka messages.
 """
 import logging
+import time
+
+import numpy.typing as npt
 
 import numpy as np
 from confluent_kafka import TopicPartition, Message, Consumer
@@ -23,7 +26,6 @@ def handle_event_messages(event_messages: list[Message], data: Data) -> None:
         event_messages: Messages received from Kafka event topic.
         data: Data served by ``kafka_dae_diagnostics``.
     """
-    logger.debug("Processing %s event messages", len(event_messages))
     for msg in event_messages:
         if msg.error():
             logger.warning("Kafka message error: %s", msg.error().code())
@@ -51,14 +53,22 @@ def handle_ev44(data: Data, msg: bytes):
         msg: Message bytes received from Kafka.
     """
     ev44 = deserialise_ev44(msg)
-
     bin_events_into_spectrum(
         histogram=data.spectra[0],
         event_tofs=ev44.time_of_flight,
         pixel_ids=ev44.pixel_id,
-        tof_bin_boundaries=np.linspace(0, 20_000_000, 1_000, dtype=np.int32)
+        tof_bin_boundaries=data.bin_boundaries
     )
 
+    data.total_events += ev44.pixel_id.size
+    data.total_event_messages += 1
+    data.total_event_megabytes += len(msg) / 1024**2
+
+    ev44_timestamp_s = ev44.reference_time[0] / 1_000_000_000
+    data.largest_kafka_timestamp = max(data.largest_kafka_timestamp, ev44_timestamp_s)
+    data.most_recent_kafka_timestamp = ev44_timestamp_s
+    data.processing_lag = max(time.time() - ev44_timestamp_s, 0)
+
 
 def handle_run_info_messages(run_info_messages: list[Message], data: Data, event_consumer: Consumer):
     """
@@ -103,20 +113,31 @@ def handle_pl72(data: Data, msg: bytes, event_consumer: Consumer) -> None:
         event_consumer: Kafka event topic consumer.
     """
     pl72 = deserialise_pl72(msg)
-    logger.info(f"Run start (filename='%s', start_time='%i', run_name='%s', instrument-name='%s', n_spectra='%i')", pl72.filename, pl72.start_time, pl72.run_name, pl72.instrument_name, pl72.detector_spectrum_map.n_spectra)
+    logger.info("Run start (filename='%s', start_time='%i', run_name='%s', instrument-name='%s', n_spectra='%i')", pl72.filename, pl72.start_time, pl72.run_name, pl72.instrument_name, pl72.detector_spectrum_map.n_spectra)
     periods = 1  # TODO
     detectors = pl72.detector_spectrum_map.n_spectra
     time_channels =  1000  # TODO
 
+    data.time_bin_boundaries = np.linspace(0, 20_000_000, time_channels+1, dtype=np.int32)  # TODO
+
     # Only reallocate if shape has changed - otherwise zero existing array.
-    if data.spectra.shape == [periods, detectors, time_channels]:
+    if data.spectra.shape == (periods, detectors, time_channels):
         data.spectra[...] = 0
     else:
         del data.spectra
-        data.spectra = np.zeros([periods, detectors, time_channels], dtype=np.uint64)
+        data.spectra = np.zeros([periods, detectors, time_channels], dtype=np.float64)
 
     # Assign event consumer to start at the time the run started
     tp = event_consumer.assignment()[0]
     event_consumer.seek(event_consumer.offsets_for_times(
         [TopicPartition(tp.topic, tp.partition, pl72.start_time)]
     )[0])
+
+    pl72_timestamp_s = pl72.start_time / 1000
+
+    data.total_events = 0
+    data.total_event_messages = 0
+    data.total_event_megabytes = 0
+    data.largest_kafka_timestamp = pl72_timestamp_s
+    data.most_recent_kafka_timestamp = pl72_timestamp_s
+    data.start_time = pl72_timestamp_s