howl MORE

Author: Tom-Willemsen

README.md

@@ -59,7 +59,10 @@ INFO:saluki:            0 - low:7515, high:7551, num_messages:36
 
 `saluki play mybroker:9092/source_topic mybroker:9092/dest_topic -t 1762209990 1762209992` - This will forward messages between the two given timestamps.
 
-## `howl` - Produce fake `ev44` messages to a topic
+## `howl` - Produce fake run-like messages
+
+`saluki-howl` emits `ev44` events, `pl72` run starts, and `6s4t` run stops to Kafka, in a format which
+look somewhat like a real run.
 
 ```
 saluki howl mybroker:9092/dest_topic --events-per-frame 200 --frames-per-second 50 --tof-peak 10000000 --tof-sigma 5000000 --det-min 0 --det-max 500

src/saluki/howl.py

@@ -1,24 +1,27 @@
+import json
 import logging
 import time
+import uuid
 
 import numpy as np
 from confluent_kafka import Producer
-from streaming_data_types import serialise_ev44
+from streaming_data_types import serialise_6s4t, serialise_ev44, serialise_pl72
+from streaming_data_types.run_start_pl72 import DetectorSpectrumMap
 
 logger = logging.getLogger("saluki")
 
 RNG = np.random.default_rng()
 
 
-def generate_fake_ev44(
+def generate_fake_events(
     msg_id: int,
     events_per_frame: int,
     tof_peak: float,
     tof_sigma: float,
     det_min: int,
     det_max: int,
 ) -> bytes:
-    detector_ids = np.random.randint(low=det_min, high=det_max, size=events_per_frame)
+    detector_ids = RNG.integers(low=det_min, high=det_max, size=events_per_frame)
     tofs = np.maximum(0.0, RNG.normal(loc=tof_peak, scale=tof_sigma, size=events_per_frame))
 
     return serialise_ev44(
@@ -31,50 +34,98 @@ def generate_fake_ev44(
     )
 
 
+def generate_run_start(det_max: int) -> bytes:
+    det_spec_map = DetectorSpectrumMap(
+        detector_ids=np.arange(0, det_max, dtype=np.int32),
+        spectrum_numbers=np.arange(0, det_max, dtype=np.int32),
+        n_spectra=det_max,
+    )
+    return serialise_pl72(
+        start_time=int(time.time() * 1000),
+        stop_time=None,
+        run_name=f"saluki-howl-{uuid.uuid4()}",
+        instrument_name="saluki-howl",
+        nexus_structure=json.dumps({}),
+        job_id=str(uuid.uuid4()),
+        filename=str(uuid.uuid4()),
+        detector_spectrum_map=det_spec_map,
+    )
+
+
+def generate_run_stop() -> bytes:
+    return serialise_6s4t(
+        stop_time=int(time.time() * 1000),
+        job_id=str(uuid.uuid4()),
+    )
+
+
 def howl(
     broker: str,
-    topic: str,
+    topic_prefix: str,
     events_per_frame: int,
     frames_per_second: int,
+    frames_per_run: int,
     tof_peak: float,
     tof_sigma: float,
     det_min: int,
     det_max: int,
 ) -> None:
     """
-    Prints the broker and topic metadata for a given broker.
-    If a topic is given, only this topic's partitions and watermarks will be printed.
-    :param broker: The broker address including port number.
-    :param topic: Optional topic to filter information to.
+    Send messages vaguely resembling a run to Kafka.
     """
 
     producer = Producer(
         {
             "bootstrap.servers": broker,
+            "queue.buffering.max.messages": 100000,
+            "queue.buffering.max.ms": 20,
         }
     )
 
     target_frame_time = 1 / frames_per_second
 
-    msg_id = 0
+    frames = 0
 
-    ev44_size = len(generate_fake_ev44(0, events_per_frame, tof_peak, tof_sigma, det_min, det_max))
+    ev44_size = len(
+        generate_fake_events(0, events_per_frame, tof_peak, tof_sigma, det_min, det_max)
+    )
     rate_bytes_per_sec = ev44_size * frames_per_second
     rate_mbit_per_sec = (rate_bytes_per_sec / 1024**2) * 8
-    logger.info(f"Attempting to simulate data rate: {rate_mbit_per_sec:.3f} MBit/s")
+    logger.info(f"Attempting to simulate data rate: {rate_mbit_per_sec:.3f} Mbit/s")
+    logger.info(f"Each ev44 is {ev44_size} bytes")
+
+    producer.produce(
+        topic=f"{topic_prefix}_runInfo",
+        key=None,
+        value=generate_run_start(det_max),
+    )
 
     while True:
         start_time = time.time()
         target_end_time = start_time + target_frame_time
 
         producer.produce(
-            topic=topic,
+            topic=f"{topic_prefix}_events",
             key=None,
-            value=generate_fake_ev44(
-                msg_id, events_per_frame, tof_peak, tof_sigma, det_min, det_max
+            value=generate_fake_events(
+                frames, events_per_frame, tof_peak, tof_sigma, det_min, det_max
             ),
         )
-        msg_id += 1
+        producer.poll(0)
+        frames += 1
+
+        if frames_per_run != 0 and frames % frames_per_run == 0:
+            logger.info(f"Starting new run after {frames_per_run} simulated frames")
+            producer.produce(
+                topic=f"{topic_prefix}_runInfo",
+                key=None,
+                value=generate_run_stop(),
+            )
+            producer.produce(
+                topic=f"{topic_prefix}_runInfo",
+                key=None,
+                value=generate_run_start(det_max),
+            )
 
         sleep_time = max(target_end_time - time.time(), 0)
         if sleep_time == 0:

src/saluki/main.py

@@ -3,10 +3,10 @@
 import sys
 
 from saluki.consume import consume
+from saluki.howl import howl
 from saluki.listen import listen
 from saluki.play import play
 from saluki.sniff import sniff
-from saluki.howl import howl
 from saluki.utils import dateutil_parsable_or_unix_timestamp, parse_kafka_uri
 
 logger = logging.getLogger("saluki")
@@ -131,13 +131,28 @@ def main() -> None:
         help="replay mode - replay data into another topic",
         parents=[common_options],
     )
-    howl_parser.add_argument("topic", type=str, help="Destination topic")
-    howl_parser.add_argument("--events-per-frame", type=int, help="Events per frame to simulate")
-    howl_parser.add_argument("--frames-per-second", type=int, help="Frames per second to simulate")
-    howl_parser.add_argument("--tof-peak", type=float, help="Time-of-flight peak (ns)")
-    howl_parser.add_argument("--tof-sigma", type=float, help="Time-of-flight sigma (ns)")
-    howl_parser.add_argument("--det-min", type=int, help="Minimum detector ID")
-    howl_parser.add_argument("--det-max", type=int, help="Maximum detector ID")
+    howl_parser.add_argument("broker", type=str, help="Kafka broker URL")
+    howl_parser.add_argument("topic_prefix", type=str, help="Topic prefix e.g. INSTNAME")
+    howl_parser.add_argument(
+        "--events-per-frame", type=int, help="Events per frame to simulate", default=100
+    )
+    howl_parser.add_argument(
+        "--frames-per-second", type=int, help="Frames per second to simulate", default=1
+    )
+    howl_parser.add_argument(
+        "--frames-per-run",
+        type=int,
+        help="Frames to take before beginning new run (0 to run forever)",
+        default=0,
+    )
+    howl_parser.add_argument(
+        "--tof-peak", type=float, help="Time-of-flight peak (ns)", default=10_000_000
+    )
+    howl_parser.add_argument(
+        "--tof-sigma", type=float, help="Time-of-flight sigma (ns)", default=2_000_000
+    )
+    howl_parser.add_argument("--det-min", type=int, help="Minimum detector ID", default=0)
+    howl_parser.add_argument("--det-max", type=int, help="Maximum detector ID", default=1000)
 
     if len(sys.argv) == 1:
         parser.print_help()
@@ -189,13 +204,12 @@ def main() -> None:
             logger.debug(f"Sniffing whole broker {args.broker}")
             sniff(args.broker)
     elif args.command == _HOWL:
-        broker, topic = parse_kafka_uri(args.topic)
-        logger.debug(f"Howling to topic {topic} on broker {broker}")
         howl(
-            broker,
-            topic,
+            args.broker,
+            args.topic_prefix,
             events_per_frame=args.events_per_frame,
             frames_per_second=args.frames_per_second,
+            frames_per_run=args.frames_per_run,
             tof_peak=args.tof_peak,
             tof_sigma=args.tof_sigma,
             det_min=args.det_min,

src/saluki/play.py

@@ -70,7 +70,8 @@ def play(
         logger.debug(f"finished consuming {num_messages} messages")
         consumer.close()
         producer.produce_batch(
-            dest_topic, [{"key": message.key(), "value": message.value()} for message in msgs]
+            dest_topic,
+            [{"key": message.key(), "value": message.value()} for message in msgs],
         )
         logger.debug(f"flushing producer. len(p): {len(producer)}")
         producer.flush(timeout=10)

src/saluki/utils.py

@@ -34,7 +34,9 @@ def fallback_deserialiser(payload: bytes) -> str:
 
 
 def deserialise_and_print_messages(
-    msgs: List[Message], partition: int | None, schemas_to_filter_to: list[str] | None = None
+    msgs: List[Message],
+    partition: int | None,
+    schemas_to_filter_to: list[str] | None = None,
 ) -> None:
     for msg in msgs:
         try:

tests/test_howl.py

@@ -78,7 +78,9 @@ def test_deserialising_message_which_raises_does_not_stop_loop(mock_message):
         assert logger.info.call_count == 1
 
 
-def test_deserialising_with_schema_list_ignores_messages_with_schema_not_in_list(mock_message):
+def test_deserialising_with_schema_list_ignores_messages_with_schema_not_in_list(
+    mock_message,
+):
     with patch("saluki.utils.logger") as logger:
         ok_message = Mock(spec=Message)
         ok_message.value.return_value = serialise_fc00(config_change=1, streams=[])  # type: ignore
@@ -179,7 +181,8 @@ def test_uri_with_no_topic():
 
 
 @pytest.mark.parametrize(
-    "timestamp", ["2025-11-19T15:27:11", "2025-11-19T15:27:11Z", "2025-11-19T15:27:11+00:00"]
+    "timestamp",
+    ["2025-11-19T15:27:11", "2025-11-19T15:27:11Z", "2025-11-19T15:27:11+00:00"],
 )
 def test_parses_datetime_properly_with_string(timestamp):
     assert dateutil_parsable_or_unix_timestamp(timestamp) == 1763566031000