DUNE-DAQ
DUNE Trigger and Data Acquisition software
Loading...
Searching...
No Matches
DAPHNEStream_binding_test Namespace Reference

Functions

int test_construction_and_size ()
int test_static_members ()
int test_bytes_roundtrip ()
int test_less_than_operator ()
int test_timestamp ()
int test_channel ()
int test_daqheader_accessible ()
int test_header_properties ()
int test_adc_single ()
int test_adc_max_value ()
int test_adc_independence ()
int test_adc_all_channels_one_sample ()
int test_adc_out_of_range ()
int main ()

Variables

int NUM_CHANNELS = 4
int NUM_ADCS = 64
tuple ADC_MAX = (1 << 14) - 1

Function Documentation

◆ main()

int main ( )

Definition at line 220 of file DAPHNEStream_binding_test.py.

220def main() -> int:
221 tests = [
222 test_construction_and_size,
223 test_static_members,
224 test_bytes_roundtrip,
225 test_less_than_operator,
226 test_timestamp,
227 test_channel,
228 test_daqheader_accessible,
229 test_header_properties,
230 test_adc_single,
231 test_adc_max_value,
232 test_adc_independence,
233 test_adc_all_channels_one_sample,
234 test_adc_out_of_range,
235 ]
236
237 failures = 0
238 for test in tests:
239 failures += test()
240
241 print()
242 if failures == 0:
243 print("All DAPHNEStreamFrame Python binding tests passed")
244 else:
245 print(f"{failures} DAPHNEStreamFrame Python binding test(s) FAILED")
246 return failures
247
248
int main(int argc, char *argv[])

◆ test_adc_all_channels_one_sample()

int test_adc_all_channels_one_sample ( )

Definition at line 187 of file DAPHNEStream_binding_test.py.

187def test_adc_all_channels_one_sample() -> int:
188 frame = DAPHNEStreamFrame()
189 adc_index = 0
190 for channel_idx in range(NUM_CHANNELS):
191 frame.set_adc(adc_index, channel_idx, (channel_idx + 1) * 100)
192 for channel_idx in range(NUM_CHANNELS):
193 expected = (channel_idx + 1) * 100
194 got = frame.get_adc(adc_index, channel_idx)
195 if got != expected:
196 print(f"FAIL: all-channels test: channel {channel_idx}: expected {expected}, got {got}")
197 return 1
198 print(f"PASS: all {NUM_CHANNELS} channels read back correctly for adc index {adc_index}")
199 return 0
200
201

◆ test_adc_independence()

int test_adc_independence ( )

Definition at line 168 of file DAPHNEStream_binding_test.py.

168def test_adc_independence() -> int:
169 frame = DAPHNEStreamFrame()
170 test_cases = [
171 (0, 0, 100),
172 (1, 0, 200),
173 (0, 1, 300),
174 (NUM_ADCS - 1, NUM_CHANNELS - 1, 9999),
175 ]
176 for adc_idx, channel_idx, val in test_cases:
177 frame.set_adc(adc_idx, channel_idx, val)
178 for adc_idx, channel_idx, expected in test_cases:
179 got = frame.get_adc(adc_idx, channel_idx)
180 if got != expected:
181 print(f"FAIL: ADC independence: adc={adc_idx}, channel={channel_idx}: expected {expected}, got {got}")
182 return 1
183 print("PASS: ADC cell independence across multiple (adc, channel) pairs")
184 return 0
185
186

◆ test_adc_max_value()

int test_adc_max_value ( )

Definition at line 157 of file DAPHNEStream_binding_test.py.

157def test_adc_max_value() -> int:
158 frame = DAPHNEStreamFrame()
159 frame.set_adc(0, 0, ADC_MAX)
160 got = frame.get_adc(0, 0)
161 if got != ADC_MAX:
162 print(f"FAIL: max ADC value {ADC_MAX}: got {got}")
163 return 1
164 print(f"PASS: max ADC value ({ADC_MAX}) preserved")
165 return 0
166
167

◆ test_adc_out_of_range()

int test_adc_out_of_range ( )

Definition at line 202 of file DAPHNEStream_binding_test.py.

202def test_adc_out_of_range() -> int:
203 frame = DAPHNEStreamFrame()
204 try:
205 frame.get_adc(NUM_ADCS, 0)
206 print("FAIL: get_adc with out-of-range adc index should have raised an exception")
207 return 1
208 except Exception:
209 pass
210 try:
211 frame.get_adc(0, NUM_CHANNELS)
212 print("FAIL: get_adc with out-of-range channel index should have raised an exception")
213 return 1
214 except Exception:
215 pass
216 print("PASS: out-of-range channel/adc indices raise exceptions as expected")
217 return 0
218
219

◆ test_adc_single()

int test_adc_single ( )

Definition at line 146 of file DAPHNEStream_binding_test.py.

146def test_adc_single() -> int:
147 frame = DAPHNEStreamFrame()
148 frame.set_adc(0, 0, 100)
149 got = frame.get_adc(0, 0)
150 if got != 100:
151 print(f"FAIL: set_adc(0,0,100) -> get_adc(0,0) returned {got}")
152 return 1
153 print("PASS: single ADC set/get (adc index 0, channel 0)")
154 return 0
155
156

◆ test_bytes_roundtrip()

int test_bytes_roundtrip ( )

Definition at line 54 of file DAPHNEStream_binding_test.py.

54def test_bytes_roundtrip() -> int:
55 frame = DAPHNEStreamFrame()
56 frame.set_timestamp(0x1234ABCD)
57 frame.header.channel_0 = 7
58 frame.header.channel_1 = 8
59 raw = frame.get_bytes()
60 if len(raw) != DAPHNEStreamFrame.sizeof():
61 print(f"FAIL: get_bytes() length {len(raw)} != sizeof() {DAPHNEStreamFrame.sizeof()}")
62 return 1
63
64 clone = DAPHNEStreamFrame(raw)
65 if clone.get_timestamp() != 0x1234ABCD:
66 print(f"FAIL: bytes round-trip timestamp mismatch: got {clone.get_timestamp():#x}")
67 return 1
68 if (clone.get_channel0(), clone.get_channel1()) != (7, 8):
69 print("FAIL: bytes round-trip channel mismatch")
70 return 1
71
72 print("PASS: DAPHNEStreamFrame bytes constructor roundtrip")
73 return 0
74
75

◆ test_channel()

int test_channel ( )

Definition at line 103 of file DAPHNEStream_binding_test.py.

103def test_channel() -> int:
104 frame = DAPHNEStreamFrame()
105 hdr = frame.header
106 hdr.channel_0 = 30
107 hdr.channel_1 = 31
108 hdr.channel_2 = 32
109 hdr.channel_3 = 33
110
111 if (frame.get_channel0(), frame.get_channel1(), frame.get_channel2(), frame.get_channel3()) != (30, 31, 32, 33):
112 print("FAIL: get_channel0..3 mismatch")
113 return 1
114
115 print("PASS: channel_0..3 header properties and get_channel0..3")
116 return 0
117
118

◆ test_construction_and_size()

int test_construction_and_size ( )

Definition at line 12 of file DAPHNEStream_binding_test.py.

12def test_construction_and_size() -> int:
13 frame = DAPHNEStreamFrame()
14 if frame is None:
15 print("FAIL: DAPHNEStreamFrame default construction returned None")
16 return 1
17 size = DAPHNEStreamFrame.sizeof()
18 if size <= 0:
19 print(f"FAIL: DAPHNEStreamFrame.sizeof() returned {size}, expected > 0")
20 return 1
21 print(f"PASS: DAPHNEStreamFrame construction and sizeof={size}")
22 return 0
23
24

◆ test_daqheader_accessible()

int test_daqheader_accessible ( )

Definition at line 119 of file DAPHNEStream_binding_test.py.

119def test_daqheader_accessible() -> int:
120 frame = DAPHNEStreamFrame()
121 hdr = frame.daq_header
122 if hdr is None:
123 print("FAIL: daq_header returned None")
124 return 1
125 print("PASS: daq_header accessible")
126 return 0
127
128

◆ test_header_properties()

int test_header_properties ( )

Definition at line 129 of file DAPHNEStream_binding_test.py.

129def test_header_properties() -> int:
130 frame = DAPHNEStreamFrame()
131 hdr = frame.header
132
133 hdr.channel_0 = 10
134 hdr.channel_1 = 11
135 hdr.channel_2 = 12
136 hdr.channel_3 = 13
137
138 if (hdr.channel_0, hdr.channel_1, hdr.channel_2, hdr.channel_3) != (10, 11, 12, 13):
139 print("FAIL: DAPHNEStreamHeader channel property mismatch")
140 return 1
141
142 print("PASS: DAPHNEStreamHeader properties")
143 return 0
144
145

◆ test_less_than_operator()

int test_less_than_operator ( )

Definition at line 76 of file DAPHNEStream_binding_test.py.

76def test_less_than_operator() -> int:
77 earlier = DAPHNEStreamFrame()
78 later = DAPHNEStreamFrame()
79 earlier.set_timestamp(100)
80 later.set_timestamp(200)
81 if not (earlier < later):
82 print("FAIL: __lt__ expected earlier < later")
83 return 1
84 if later < earlier:
85 print("FAIL: __lt__ expected later !< earlier")
86 return 1
87 print("PASS: __lt__ ordering by timestamp")
88 return 0
89
90

◆ test_static_members()

int test_static_members ( )

Definition at line 25 of file DAPHNEStream_binding_test.py.

25def test_static_members() -> int:
26 if DAPHNEStreamFrame.s_channels_per_frame != NUM_CHANNELS:
27 print(
28 "FAIL: DAPHNEStreamFrame.s_channels_per_frame expected "
29 f"{NUM_CHANNELS}, got {DAPHNEStreamFrame.s_channels_per_frame}"
30 )
31 return 1
32 if DAPHNEStreamFrame.s_adcs_per_channel != NUM_ADCS:
33 print(
34 "FAIL: DAPHNEStreamFrame.s_adcs_per_channel expected "
35 f"{NUM_ADCS}, got {DAPHNEStreamFrame.s_adcs_per_channel}"
36 )
37 return 1
38 if DAPHNEStreamFrame.s_bits_per_adc != 14:
39 print(f"FAIL: DAPHNEStreamFrame.s_bits_per_adc expected 14, got {DAPHNEStreamFrame.s_bits_per_adc}")
40 return 1
41 if DAPHNEStreamFrame.s_expected_bytes != DAPHNEStreamFrame.sizeof():
42 print(
43 "FAIL: DAPHNEStreamFrame.s_expected_bytes "
44 f"{DAPHNEStreamFrame.s_expected_bytes} != sizeof {DAPHNEStreamFrame.sizeof()}"
45 )
46 return 1
47 if DAPHNEStreamHeader.s_expected_bytes <= 0:
48 print(f"FAIL: DAPHNEStreamHeader.s_expected_bytes invalid: {DAPHNEStreamHeader.s_expected_bytes}")
49 return 1
50 print("PASS: DAPHNEStreamFrame/DAPHNEStreamHeader static members")
51 return 0
52
53

◆ test_timestamp()

int test_timestamp ( )

Definition at line 91 of file DAPHNEStream_binding_test.py.

91def test_timestamp() -> int:
92 frame = DAPHNEStreamFrame()
93 test_ts = 0x998877665544
94 frame.set_timestamp(test_ts)
95 got = frame.get_timestamp()
96 if got != test_ts:
97 print(f"FAIL: set/get_timestamp: set {test_ts:#x}, got {got:#x}")
98 return 1
99 print("PASS: set_timestamp / get_timestamp")
100 return 0
101
102

Variable Documentation

◆ ADC_MAX

tuple DAPHNEStream_binding_test.ADC_MAX = (1 << 14) - 1

Definition at line 9 of file DAPHNEStream_binding_test.py.

◆ NUM_ADCS

int DAPHNEStream_binding_test.NUM_ADCS = 64

Definition at line 8 of file DAPHNEStream_binding_test.py.

◆ NUM_CHANNELS

int DAPHNEStream_binding_test.NUM_CHANNELS = 4

Definition at line 7 of file DAPHNEStream_binding_test.py.