new forwarder config schema (#98)

* add fc00 schema for forwarder configuration

* format

Author: hurvan

README.md

@@ -22,7 +22,8 @@ https://github.com/ess-dmsc/streaming-data-types
 | tdct | Timestamps                                                                   |
 | ep00 | EPICS connection info (deprecated in favour of ep01)                         |
 | ep01 | EPICS connection info                                                        |
-| rf5k | Forwarder configuration update                                               |
+| rf5k | Forwarder configuration update (deprecated in favour of fc00)                |
+| fc00 | Forwarder configuration update                                               |
 | answ | File-writer command response                                                 |
 | wrdn | File-writer finished writing                                                 |
 | NDAr | **Deprecated**                                                               |
@@ -31,7 +32,7 @@ https://github.com/ess-dmsc/streaming-data-types
 | senv | **Deprecated**                                                               |
 | json | Generic JSON data                                                            |
 | se00 | Arrays with optional timestamps, for example waveform data. Replaces _senv_. |
-| da00 | Scipp-like data arrays, for histograms, etc.                                  |
+| da00 | Scipp-like data arrays, for histograms, etc.                                 |
 
 ### hs00 and hs01
 Schema for histogram data. It is one of the more complicated to use schemas.

streaming_data_types/__init__.py

@@ -19,6 +19,10 @@
     deserialise_rf5k,
     serialise_rf5k,
 )
+from streaming_data_types.forwarder_config_update_fc00 import (
+    deserialise_fc00,
+    serialise_fc00,
+)
 from streaming_data_types.histogram_hs00 import deserialise_hs00, serialise_hs00
 from streaming_data_types.histogram_hs01 import deserialise_hs01, serialise_hs01
 from streaming_data_types.json_json import deserialise_json, serialise_json
@@ -52,6 +56,7 @@
     "ep01": serialise_ep01,
     "tdct": serialise_tdct,
     "rf5k": serialise_rf5k,
+    "fc00": serialise_fc00,
     "answ": serialise_answ,
     "wrdn": serialise_wrdn,
     "NDAr": serialise_ndar,
@@ -81,6 +86,7 @@
     "ep01": deserialise_ep01,
     "tdct": deserialise_tdct,
     "rf5k": deserialise_rf5k,
+    "fc00": deserialise_fc00,
     "answ": deserialise_answ,
     "wrdn": deserialise_wrdn,
     "NDAr": deserialise_ndar,

streaming_data_types/fbschemas/forwarder_config_update_fc00/Protocol.py

@@ -0,0 +1,8 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: 
+
+class Protocol(object):
+    PVA = 0
+    CA = 1
+    FAKE = 2

streaming_data_types/fbschemas/forwarder_config_update_fc00/Stream.py

@@ -0,0 +1,86 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: 
+
+import flatbuffers
+from flatbuffers.compat import import_numpy
+np = import_numpy()
+
+class Stream(object):
+    __slots__ = ['_tab']
+
+    @classmethod
+    def GetRootAs(cls, buf, offset=0):
+        n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
+        x = Stream()
+        x.Init(buf, n + offset)
+        return x
+
+    @classmethod
+    def GetRootAsStream(cls, buf, offset=0):
+        """This method is deprecated. Please switch to GetRootAs."""
+        return cls.GetRootAs(buf, offset)
+    @classmethod
+    def StreamBufferHasIdentifier(cls, buf, offset, size_prefixed=False):
+        return flatbuffers.util.BufferHasIdentifier(buf, offset, b"\x66\x63\x30\x30", size_prefixed=size_prefixed)
+
+    # Stream
+    def Init(self, buf, pos):
+        self._tab = flatbuffers.table.Table(buf, pos)
+
+    # Stream
+    def Channel(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
+        if o != 0:
+            return self._tab.String(o + self._tab.Pos)
+        return None
+
+    # Stream
+    def Schema(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+        if o != 0:
+            return self._tab.String(o + self._tab.Pos)
+        return None
+
+    # Stream
+    def Topic(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(8))
+        if o != 0:
+            return self._tab.String(o + self._tab.Pos)
+        return None
+
+    # Stream
+    def Protocol(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
+        if o != 0:
+            return self._tab.Get(flatbuffers.number_types.Uint16Flags, o + self._tab.Pos)
+        return 0
+
+    # Stream
+    def Periodic(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(12))
+        if o != 0:
+            return self._tab.Get(flatbuffers.number_types.Int32Flags, o + self._tab.Pos)
+        return 0
+
+def StreamStart(builder): builder.StartObject(5)
+def Start(builder):
+    return StreamStart(builder)
+def StreamAddChannel(builder, channel): builder.PrependUOffsetTRelativeSlot(0, flatbuffers.number_types.UOffsetTFlags.py_type(channel), 0)
+def AddChannel(builder, channel):
+    return StreamAddChannel(builder, channel)
+def StreamAddSchema(builder, schema): builder.PrependUOffsetTRelativeSlot(1, flatbuffers.number_types.UOffsetTFlags.py_type(schema), 0)
+def AddSchema(builder, schema):
+    return StreamAddSchema(builder, schema)
+def StreamAddTopic(builder, topic): builder.PrependUOffsetTRelativeSlot(2, flatbuffers.number_types.UOffsetTFlags.py_type(topic), 0)
+def AddTopic(builder, topic):
+    return StreamAddTopic(builder, topic)
+def StreamAddProtocol(builder, protocol): builder.PrependUint16Slot(3, protocol, 0)
+def AddProtocol(builder, protocol):
+    return StreamAddProtocol(builder, protocol)
+def StreamAddPeriodic(builder, periodic): builder.PrependInt32Slot(4, periodic, 0)
+def AddPeriodic(builder, periodic):
+    return StreamAddPeriodic(builder, periodic)
+def StreamEnd(builder): return builder.EndObject()
+def End(builder):
+    return StreamEnd(builder)

streaming_data_types/fbschemas/forwarder_config_update_fc00/UpdateType.py

@@ -0,0 +1,9 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: 
+
+class UpdateType(object):
+    ADD = 0
+    REMOVE = 1
+    REMOVEALL = 2
+    REPLACE = 3

streaming_data_types/fbschemas/forwarder_config_update_fc00/__init__.py

@@ -0,0 +1,77 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: 
+
+import flatbuffers
+from flatbuffers.compat import import_numpy
+np = import_numpy()
+
+class fc00_ConfigUpdate(object):
+    __slots__ = ['_tab']
+
+    @classmethod
+    def GetRootAs(cls, buf, offset=0):
+        n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
+        x = fc00_ConfigUpdate()
+        x.Init(buf, n + offset)
+        return x
+
+    @classmethod
+    def GetRootAsfc00_ConfigUpdate(cls, buf, offset=0):
+        """This method is deprecated. Please switch to GetRootAs."""
+        return cls.GetRootAs(buf, offset)
+    @classmethod
+    def fc00_ConfigUpdateBufferHasIdentifier(cls, buf, offset, size_prefixed=False):
+        return flatbuffers.util.BufferHasIdentifier(buf, offset, b"\x66\x63\x30\x30", size_prefixed=size_prefixed)
+
+    # fc00_ConfigUpdate
+    def Init(self, buf, pos):
+        self._tab = flatbuffers.table.Table(buf, pos)
+
+    # fc00_ConfigUpdate
+    def ConfigChange(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
+        if o != 0:
+            return self._tab.Get(flatbuffers.number_types.Uint16Flags, o + self._tab.Pos)
+        return 0
+
+    # fc00_ConfigUpdate
+    def Streams(self, j):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+        if o != 0:
+            x = self._tab.Vector(o)
+            x += flatbuffers.number_types.UOffsetTFlags.py_type(j) * 4
+            x = self._tab.Indirect(x)
+            from .Stream import Stream
+            obj = Stream()
+            obj.Init(self._tab.Bytes, x)
+            return obj
+        return None
+
+    # fc00_ConfigUpdate
+    def StreamsLength(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+        if o != 0:
+            return self._tab.VectorLen(o)
+        return 0
+
+    # fc00_ConfigUpdate
+    def StreamsIsNone(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+        return o == 0
+
+def fc00_ConfigUpdateStart(builder): builder.StartObject(2)
+def Start(builder):
+    return fc00_ConfigUpdateStart(builder)
+def fc00_ConfigUpdateAddConfigChange(builder, configChange): builder.PrependUint16Slot(0, configChange, 0)
+def AddConfigChange(builder, configChange):
+    return fc00_ConfigUpdateAddConfigChange(builder, configChange)
+def fc00_ConfigUpdateAddStreams(builder, streams): builder.PrependUOffsetTRelativeSlot(1, flatbuffers.number_types.UOffsetTFlags.py_type(streams), 0)
+def AddStreams(builder, streams):
+    return fc00_ConfigUpdateAddStreams(builder, streams)
+def fc00_ConfigUpdateStartStreamsVector(builder, numElems): return builder.StartVector(4, numElems, 4)
+def StartStreamsVector(builder, numElems):
+    return fc00_ConfigUpdateStartStreamsVector(builder, numElems)
+def fc00_ConfigUpdateEnd(builder): return builder.EndObject()
+def End(builder):
+    return fc00_ConfigUpdateEnd(builder)

streaming_data_types/fbschemas/forwarder_config_update_fc00/fc00_ConfigUpdate.py

@@ -0,0 +1,119 @@
+from collections import namedtuple
+from typing import List, Union
+
+import flatbuffers
+from flatbuffers.packer import struct as flatbuffer_struct
+
+from streaming_data_types.fbschemas.forwarder_config_update_fc00 import (
+    Protocol,
+    Stream,
+    UpdateType,
+    fc00_ConfigUpdate,
+)
+from streaming_data_types.utils import check_schema_identifier
+
+FILE_IDENTIFIER = b"fc00"
+
+ConfigurationUpdate = namedtuple("ConfigurationUpdate", ("config_change", "streams"))
+
+StreamInfo = namedtuple(
+    "StreamInfo", ("channel", "schema", "topic", "protocol", "periodic")
+)
+
+
+def deserialise_fc00(buffer: Union[bytearray, bytes]) -> ConfigurationUpdate:
+    """
+    Deserialise FlatBuffer fc00.
+
+    :param buffer: The FlatBuffers buffer.
+    :return: The deserialised data.
+    """
+    check_schema_identifier(buffer, FILE_IDENTIFIER)
+
+    config_message = fc00_ConfigUpdate.fc00_ConfigUpdate.GetRootAsfc00_ConfigUpdate(
+        buffer, 0
+    )
+
+    streams = []
+    try:
+        for i in range(config_message.StreamsLength()):
+            stream_message = config_message.Streams(i)
+            streams.append(
+                StreamInfo(
+                    stream_message.Channel().decode("utf-8")
+                    if stream_message.Channel()
+                    else "",
+                    stream_message.Schema().decode("utf-8")
+                    if stream_message.Schema()
+                    else "",
+                    stream_message.Topic().decode("utf-8")
+                    if stream_message.Topic()
+                    else "",
+                    stream_message.Protocol(),
+                    int(stream_message.Periodic().decode("utf-8"))
+                    if stream_message.Periodic()
+                    else 0,
+                )
+            )
+    except flatbuffer_struct.error:
+        pass  # No streams in buffer
+
+    return ConfigurationUpdate(config_message.ConfigChange(), streams)
+
+
+def serialise_stream(
+    builder: flatbuffers.Builder,
+    protocol: Protocol,
+    channel_offset: int,
+    schema_offset: int,
+    topic_offset: int,
+) -> int:
+    Stream.StreamStart(builder)
+    Stream.StreamAddProtocol(builder, protocol)
+    Stream.StreamAddTopic(builder, topic_offset)
+    Stream.StreamAddSchema(builder, schema_offset)
+    Stream.StreamAddChannel(builder, channel_offset)
+    return Stream.StreamEnd(builder)
+
+
+def serialise_fc00(config_change: UpdateType, streams: List[StreamInfo]) -> bytes:
+    """
+    Serialise config update message as an fc00 FlatBuffers message.
+
+    :param config_change:
+    :param streams: channel, schema and output topic configurations
+    :return:
+    """
+    builder = flatbuffers.Builder(1024)
+    builder.ForceDefaults(True)
+
+    if streams:
+        # We have to use multiple loops/list comprehensions here because we cannot create strings after we have
+        # called StreamStart and cannot create streams after we have called StartVector
+        stream_field_offsets = [
+            (
+                builder.CreateString(stream.channel),
+                builder.CreateString(stream.schema),
+                builder.CreateString(stream.topic),
+            )
+            for stream in streams
+        ]
+        stream_offsets = [
+            serialise_stream(builder, stream.protocol, *stream_fields)
+            for stream, stream_fields in zip(streams, stream_field_offsets)
+        ]
+
+        fc00_ConfigUpdate.fc00_ConfigUpdateStartStreamsVector(builder, len(streams))
+        for stream_offset in stream_offsets:
+            builder.PrependUOffsetTRelative(stream_offset)
+        streams_offset = builder.EndVector()
+
+    # Build the actual buffer
+    fc00_ConfigUpdate.fc00_ConfigUpdateStart(builder)
+    if streams:
+        fc00_ConfigUpdate.fc00_ConfigUpdateAddStreams(builder, streams_offset)
+    fc00_ConfigUpdate.fc00_ConfigUpdateAddConfigChange(builder, config_change)
+    data = fc00_ConfigUpdate.fc00_ConfigUpdateEnd(builder)
+
+    builder.Finish(data, file_identifier=FILE_IDENTIFIER)
+    return bytes(builder.Output())