Skip to content

Middlewares

Small helpers to parse the binary frames in the workers.

Contrast

parse(data)

Parses a contrast packet, which returns a dict, depending on the status of contrast

Parameters:

Name Type Description Default
data StreamData

a frame comming from contrast

 required

Returns:

Type Description
UnionType

a ContrastPacket containing a.o. the current status
Source code in dranspose/middlewares/contrast.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def parse(data: StreamData) -> UnionType:
    """
    Parses a contrast packet, which returns a dict, depending on the status of contrast

    Arguments:
        data: a frame comming from contrast

    Returns:
        a ContrastPacket containing a.o. the current status
    """
    assert data.typ == "contrast"
    assert data.length == 1
    frame = data.frames[0]
    if isinstance(frame, zmq.Frame):
        val = pickle.loads(frame.bytes)
    else:
        val = pickle.loads(frame)

    return ContrastPacket.validate_python(val)

Xspress3

parse(data)

Parses a Xspress3 packet, which either gives a start/end message or a tuple with a spectra array

Parameters:

Name Type Description Default
data StreamData

a frame comming from the Xspress3 tango device

 required

Returns:

Type Description
UnionType

an XspressPacket
Source code in dranspose/middlewares/xspress.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def parse(
    data: StreamData,
) -> UnionType:
    """
    Parses a Xspress3 packet, which either gives a start/end message or a tuple with a spectra array

    Arguments:
        data: a frame comming from the Xspress3 tango device

    Returns:
        an XspressPacket
    """
    assert data.typ == "xspress", "wrong packet typ"
    assert data.length >= 1, "wrong number of multiparts"
    headerframe = data.frames[0]
    if isinstance(headerframe, zmq.Frame):
        headerframe = headerframe.bytes
    packet = XspressPacket.validate_json(headerframe)
    print("packets", packet)
    if isinstance(packet, XspressImage):
        assert data.length == 3, "images does not contain data"
        bufframe = data.frames[1]
        if isinstance(bufframe, zmq.Frame):
            bufframe = bufframe.bytes
        buf = np.frombuffer(bufframe, dtype=packet.type)
        img = buf.reshape(packet.shape)
        metaframe = data.frames[2]
        if isinstance(metaframe, zmq.Frame):
            metaframe = metaframe.bytes
        meta = pickle.loads(metaframe)
        # meta description: ocr[0], AllEvents[0], AllGood[0], ClockTicks[0],
        #                   TotalTicks[0], ResetTicks[0], event_widths, dtc[0]]
        meta = {
            k: v
            for k, v in zip(
                [
                    "ocr",
                    "AllEvents",
                    "AllGood",
                    "ClockTicks",
                    "TotalTicks",
                    "ResetTicks",
                    "event_widths",
                    "dtc",
                ],
                meta,
            )
        }
        packet.data = img
        packet.meta = meta

    return packet