Merge pull request #19 from ess-dmsc/add_ep00_schema

Add ep00 schema

Author: mattclarke

README.md

@@ -16,6 +16,7 @@ https://github.com/ess-dmsc/streaming-data-types
 |ev42|Event Data|Y|
 |x5f2|Status messages|N|
 |tdct|Timestamps|Y|
+|ep00|EPICS Connection Info|Y|
 
 \* whether it passes verification via the C++ FlatBuffers library.
 

streaming_data_types/epics_connection_info_ep00.py

@@ -0,0 +1,63 @@
+from typing import Union, Optional
+import flatbuffers
+from streaming_data_types.fbschemas.epics_connection_info_ep00 import (
+    EpicsConnectionInfo,
+    EventType,
+)
+from streaming_data_types.utils import check_schema_identifier
+from collections import namedtuple
+
+FILE_IDENTIFIER = b"ep00"
+
+
+def serialise_ep00(
+    timestamp_ns: int,
+    event_type: EventType,
+    source_name: str,
+    service_id: Optional[str] = None,
+) -> bytes:
+    builder = flatbuffers.Builder(136)
+
+    if service_id is not None:
+        service_id_offset = builder.CreateString(service_id)
+    source_name_offset = builder.CreateString(source_name)
+
+    EpicsConnectionInfo.EpicsConnectionInfoStart(builder)
+    if service_id is not None:
+        EpicsConnectionInfo.EpicsConnectionInfoAddServiceId(builder, service_id_offset)
+    EpicsConnectionInfo.EpicsConnectionInfoAddSourceName(builder, source_name_offset)
+    EpicsConnectionInfo.EpicsConnectionInfoAddType(builder, event_type)
+    EpicsConnectionInfo.EpicsConnectionInfoAddTimestamp(builder, timestamp_ns)
+
+    end = EpicsConnectionInfo.EpicsConnectionInfoEnd(builder)
+    builder.Finish(end)
+
+    # Generate the output and replace the file_identifier
+    buffer = builder.Output()
+    buffer[4:8] = FILE_IDENTIFIER
+    return bytes(buffer)
+
+
+EpicsConnection = namedtuple(
+    "EpicsConnection", ("timestamp", "type", "source_name", "service_id",),
+)
+
+
+def deserialise_ep00(buffer: Union[bytearray, bytes]) -> EpicsConnection:
+    check_schema_identifier(buffer, FILE_IDENTIFIER)
+
+    epics_connection = EpicsConnectionInfo.EpicsConnectionInfo.GetRootAsEpicsConnectionInfo(
+        buffer, 0
+    )
+
+    source_name = (
+        epics_connection.SourceName() if epics_connection.SourceName() else b""
+    )
+    service_id = epics_connection.ServiceId() if epics_connection.ServiceId() else b""
+
+    return EpicsConnection(
+        epics_connection.Timestamp(),
+        epics_connection.Type(),
+        source_name.decode(),
+        service_id.decode(),
+    )

streaming_data_types/fbschemas/epics_connection_info_ep00/EpicsConnectionInfo.py

@@ -0,0 +1,80 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace:
+
+import flatbuffers
+
+
+class EpicsConnectionInfo(object):
+    __slots__ = ["_tab"]
+
+    @classmethod
+    def GetRootAsEpicsConnectionInfo(cls, buf, offset):
+        n = flatbuffers.encode.Get(flatbuffers.packer.uoffset, buf, offset)
+        x = EpicsConnectionInfo()
+        x.Init(buf, n + offset)
+        return x
+
+    # EpicsConnectionInfo
+    def Init(self, buf, pos):
+        self._tab = flatbuffers.table.Table(buf, pos)
+
+    # EpicsConnectionInfo
+    def Timestamp(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(4))
+        if o != 0:
+            return self._tab.Get(
+                flatbuffers.number_types.Uint64Flags, o + self._tab.Pos
+            )
+        return 0
+
+    # EpicsConnectionInfo
+    def Type(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(6))
+        if o != 0:
+            return self._tab.Get(
+                flatbuffers.number_types.Uint16Flags, o + self._tab.Pos
+            )
+        return 0
+
+    # EpicsConnectionInfo
+    def SourceName(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
+
+    # EpicsConnectionInfo
+    def ServiceId(self):
+        o = flatbuffers.number_types.UOffsetTFlags.py_type(self._tab.Offset(10))
+        if o != 0:
+            return self._tab.String(o + self._tab.Pos)
+        return None
+
+
+def EpicsConnectionInfoStart(builder):
+    builder.StartObject(4)
+
+
+def EpicsConnectionInfoAddTimestamp(builder, timestamp):
+    builder.PrependUint64Slot(0, timestamp, 0)
+
+
+def EpicsConnectionInfoAddType(builder, type):
+    builder.PrependUint16Slot(1, type, 0)
+
+
+def EpicsConnectionInfoAddSourceName(builder, sourceName):
+    builder.PrependUOffsetTRelativeSlot(
+        2, flatbuffers.number_types.UOffsetTFlags.py_type(sourceName), 0
+    )
+
+
+def EpicsConnectionInfoAddServiceId(builder, serviceId):
+    builder.PrependUOffsetTRelativeSlot(
+        3, flatbuffers.number_types.UOffsetTFlags.py_type(serviceId), 0
+    )
+
+
+def EpicsConnectionInfoEnd(builder):
+    return builder.EndObject()

streaming_data_types/fbschemas/epics_connection_info_ep00/EventType.py

@@ -0,0 +1,11 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace:
+
+
+class EventType(object):
+    UNKNOWN = 0
+    NEVER_CONNECTED = 1
+    CONNECTED = 2
+    DISCONNECTED = 3
+    DESTROYED = 4

streaming_data_types/fbschemas/epics_connection_info_ep00/__init__.py

@@ -0,0 +1,34 @@
+import pytest
+from streaming_data_types.fbschemas.epics_connection_info_ep00 import EventType
+from streaming_data_types.epics_connection_info_ep00 import (
+    serialise_ep00,
+    deserialise_ep00,
+)
+
+
+class TestSerialisationEp00:
+    original_entry = {
+        "timestamp_ns": 1593620746000000000,
+        "event_type": EventType.EventType.DISCONNECTED,
+        "source_name": "test_source",
+        "service_id": "test_service",
+    }
+
+    def test_serialises_and_deserialises_ep00_message_correctly(self):
+        buf = serialise_ep00(**self.original_entry)
+        deserialised_tuple = deserialise_ep00(buf)
+
+        assert deserialised_tuple.timestamp == self.original_entry["timestamp_ns"]
+        assert deserialised_tuple.type == self.original_entry["event_type"]
+        assert deserialised_tuple.source_name == self.original_entry["source_name"]
+        assert deserialised_tuple.service_id == self.original_entry["service_id"]
+
+    def test_if_buffer_has_wrong_id_then_throws(self):
+        buf = serialise_ep00(**self.original_entry)
+
+        # Manually hack the id
+        buf = bytearray(buf)
+        buf[4:8] = b"1234"
+
+        with pytest.raises(RuntimeError):
+            deserialise_ep00(buf)