Merge pull request #51 from ess-dmsc/add_control_topic

Added control_topic to pl72 run start

streaming_data_types/fbschemas/run_start_pl72/RunStart.py

@@ -123,9 +123,16 @@ def Metadata(self):
             return self._tab.String(o + self._tab.Pos)
         return None
 
+    # RunStart
+    def ControlTopic(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(28))
+        if o != 0:
+            return self._tab.String(o + self._tab.Pos)
+        return None
+
 
 def RunStartStart(builder):
-    builder.StartObject(12)
+    builder.StartObject(13)
 
 
 def RunStartAddStartTime(builder, startTime):
@@ -194,5 +201,11 @@ def RunStartAddMetadata(builder, metadata):
     )
 
 
+def RunStartAddControlTopic(builder, controlTopic):
+    builder.PrependUOffsetTRelativeSlot(
+        12, flatbuffers.number_types.UOffsetTFlags.py_type(controlTopic), 0
+    )
+
+
 def RunStartEnd(builder):
     return builder.EndObject()

streaming_data_types/run_start_pl72.py

@@ -36,6 +36,7 @@ def serialise_pl72(
     broker: str = "localhost:9092",
     metadata: str = "{}",
     detector_spectrum_map: Optional[DetectorSpectrumMap] = None,
+    control_topic: str = "",
 ) -> bytes:
     builder = flatbuffers.Builder(512)
     builder.ForceDefaults(True)
@@ -59,6 +60,7 @@ def serialise_pl72(
     run_name_offset = builder.CreateString(run_name)
     filename_offset = builder.CreateString(filename)
     metadata_offset = builder.CreateString(metadata)
+    control_topic_offset = builder.CreateString(control_topic)
 
     # Build detector-spectrum map
     if detector_spectrum_map is not None:
@@ -97,6 +99,7 @@ def serialise_pl72(
     RunStart.RunStartAddMetadata(builder, metadata_offset)
     if detector_spectrum_map is not None:
         RunStart.RunStartAddDetectorSpectrumMap(builder, detector_spectrum_map_offset)
+    RunStart.RunStartAddControlTopic(builder, control_topic_offset)
 
     run_start_message = RunStart.RunStartEnd(builder)
 
@@ -116,6 +119,7 @@ class RunStartInfo(NamedTuple):
     instrument_name: str = ""
     metadata: str = ""
     detector_spectrum_map: Optional[DetectorSpectrumMap] = None
+    control_topic: str = ""
 
 
 def deserialise_pl72(buffer: Union[bytearray, bytes]) -> RunStartInfo:
@@ -130,6 +134,7 @@ def deserialise_pl72(buffer: Union[bytearray, bytes]) -> RunStartInfo:
     instrument_name = run_start.InstrumentName() if run_start.InstrumentName() else b""
     run_name = run_start.RunName() if run_start.RunName() else b""
     metadata = run_start.Metadata() if run_start.Metadata() else b""
+    control_topic = run_start.ControlTopic() if run_start.ControlTopic() else b""
 
     detector_spectrum_map = None
     det_spec_map_buf = run_start.DetectorSpectrumMap()
@@ -152,4 +157,5 @@ def deserialise_pl72(buffer: Union[bytearray, bytes]) -> RunStartInfo:
         broker=broker.decode(),
         metadata=metadata.decode(),
         detector_spectrum_map=detector_spectrum_map,
+        control_topic=control_topic.decode(),
     )

tests/test_pl72.py

@@ -24,6 +24,7 @@ class TestSerialisationPl72:
         "detector_spectrum_map": DetectorSpectrumMap(
             np.array([4, 5, 6]), np.array([0, 1, 2]), 3
         ),
+        "control_topic": "some_topic_name",
     }
 
     def test_serialises_and_deserialises_pl72_message_correctly(self):
@@ -57,6 +58,7 @@ def test_serialises_and_deserialises_pl72_message_correctly(self):
             deserialised_tuple.detector_spectrum_map.detector_ids,
             self.original_entry["detector_spectrum_map"].detector_ids,
         )
+        assert deserialised_tuple.control_topic == self.original_entry["control_topic"]
 
     def test_if_buffer_has_wrong_id_then_throws(self):
         buf = serialise_pl72(**self.original_entry)