examples/a121/algo/touchless_button/processor.py#
1# Copyright (c) Acconeer AB, 2022-2026
2# All rights reserved
3
4from __future__ import annotations
5
6import numpy as np
7
8import pyqtgraph as pg
9
10import acconeer.exptool as et
11from acconeer.exptool import a121
12from acconeer.exptool.a121.algo.touchless_button import (
13 MeasurementType,
14 Processor,
15 ProcessorConfig,
16 ProcessorResult,
17 get_close_sensor_config,
18)
19
20
21def main():
22 args = a121.ExampleArgumentParser().parse_args()
23 et.utils.config_logging(args)
24
25 client = a121.Client.open(**a121.get_client_args(args))
26
27 processor_config = ProcessorConfig()
28
29 sensor_config = get_close_sensor_config()
30
31 metadata = client.setup_session(sensor_config)
32 client.start_session()
33
34 processor = Processor(
35 sensor_config=sensor_config,
36 metadata=metadata,
37 processor_config=processor_config,
38 )
39
40 pg_updater = PGUpdater(sensor_config=sensor_config, processor_config=processor_config)
41 pg_process = et.PGProcess(pg_updater)
42 pg_process.start()
43
44 interrupt_handler = et.utils.ExampleInterruptHandler()
45
46 while not interrupt_handler.got_signal:
47 result = client.get_next()
48 processor_result = processor.process(result)
49 try:
50 pg_process.put_data(processor_result)
51 except et.PGProccessDiedException:
52 break
53
54 print("Disconnecting...")
55 pg_process.close()
56 client.stop_session()
57 client.close()
58
59
60class PGUpdater:
61 def __init__(self, sensor_config, processor_config):
62 self.detection_history = np.full((2, 100), np.nan)
63 self.sensor_config = sensor_config
64 self.processor_config = processor_config
65
66 def setup(self, win):
67 self.detection_history_plot = self._create_detection_plot(win)
68
69 self.detection_history_curve_close = self.detection_history_plot.plot(
70 pen=et.utils.pg_pen_cycler(1, width=5)
71 )
72 self.detection_history_curve_far = self.detection_history_plot.plot(
73 pen=et.utils.pg_pen_cycler(0, width=5)
74 )
75
76 close_html = (
77 '<div style="text-align: center">'
78 '<span style="color: #FFFFFF;font-size:15pt;">'
79 "{}</span></div>".format("Close detection")
80 )
81 far_html = (
82 '<div style="text-align: center">'
83 '<span style="color: #FFFFFF;font-size:15pt;">'
84 "{}</span></div>".format("Far detection")
85 )
86 self.close_text_item = pg.TextItem(
87 html=close_html,
88 fill=pg.mkColor(0xFF, 0x7F, 0x0E),
89 anchor=(0.5, 0),
90 )
91 self.far_text_item = pg.TextItem(
92 html=far_html,
93 fill=pg.mkColor(0x1F, 0x77, 0xB4),
94 anchor=(0.5, 0),
95 )
96 pos_left = (100 / 3, 1.8)
97 pos_right = (2 * 100 / 3, 1.8)
98 self.close_text_item.setPos(*pos_left)
99 self.far_text_item.setPos(*pos_right)
100 self.detection_history_plot.addItem(self.close_text_item)
101 self.detection_history_plot.addItem(self.far_text_item)
102 self.close_text_item.hide()
103 self.far_text_item.hide()
104
105 self.detection_history = np.full((2, 100), np.nan)
106
107 self.score_history_plot = self._create_score_plot(win)
108 score_plot_legend = self.score_history_plot.legend
109 self.score_smooth_max = et.utils.SmoothMax()
110
111 self.threshold_history_curve_close = self.score_history_plot.plot(
112 pen=et.utils.pg_pen_cycler(1, width=2.5, style="--"),
113 )
114 self.threshold_history_curve_far = self.score_history_plot.plot(
115 pen=et.utils.pg_pen_cycler(0, width=2.5, style="--"),
116 )
117
118 self.threshold_history = np.full((2, 100), np.nan)
119
120 cycle_index = 2 # To not have same colors as thresholds
121 if self.processor_config.measurement_type == MeasurementType.CLOSE_RANGE:
122 measurement_type = "Close"
123 score_plot_legend.addItem(self.threshold_history_curve_close, "Close range threshold")
124 elif self.processor_config.measurement_type == MeasurementType.FAR_RANGE:
125 measurement_type = "Far"
126 score_plot_legend.addItem(self.threshold_history_curve_far, "Far range threshold")
127
128 if self.processor_config.measurement_type != MeasurementType.CLOSE_AND_FAR_RANGE:
129 score_history = np.full((self.sensor_config.subsweep.num_points, 100), np.nan)
130 score_history_curves = np.empty(
131 (self.sensor_config.subsweep.num_points,), dtype=object
132 )
133 for i in range(self.sensor_config.subsweep.num_points):
134 score_history_curves[i] = pg.ScatterPlotItem(
135 brush=et.utils.pg_brush_cycler(cycle_index),
136 name=f"{measurement_type} range, point {i}",
137 )
138 self.score_history_plot.addItem(score_history_curves[i])
139 cycle_index += 1
140
141 if self.processor_config.measurement_type == MeasurementType.CLOSE_RANGE:
142 self.score_history_close = score_history
143 self.score_history_curves_close = score_history_curves
144 self.score_history_far = None
145 self.score_history_curves_far = None
146 elif self.processor_config.measurement_type == MeasurementType.FAR_RANGE:
147 self.score_history_close = None
148 self.score_history_curves_close = None
149 self.score_history_far = score_history
150 self.score_history_curves_far = score_history_curves
151
152 elif self.processor_config.measurement_type == MeasurementType.CLOSE_AND_FAR_RANGE:
153 score_plot_legend.addItem(self.threshold_history_curve_close, "Close range threshold")
154 score_plot_legend.addItem(self.threshold_history_curve_far, "Far range threshold")
155 self.score_history_close = np.full(
156 (self.sensor_config.subsweeps[0].num_points, 100), np.nan
157 )
158 self.score_history_far = np.full(
159 (self.sensor_config.subsweeps[1].num_points, 100), np.nan
160 )
161 self.score_history_curves_close = np.empty(
162 (self.sensor_config.subsweeps[0].num_points,), dtype=object
163 )
164 self.score_history_curves_far = np.empty(
165 (self.sensor_config.subsweeps[1].num_points,), dtype=object
166 )
167
168 range_labels = ["Close", "Far"]
169 for n, subsweep in enumerate(self.sensor_config.subsweeps):
170 measurement_type = range_labels[n]
171 score_history_curve_list = [
172 self.score_history_curves_close,
173 self.score_history_curves_far,
174 ]
175 for i in range(subsweep.num_points):
176 score_history_curve_list[n][i] = pg.ScatterPlotItem(
177 brush=et.utils.pg_brush_cycler(cycle_index),
178 name=f"{measurement_type} range, point {i}",
179 )
180 self.score_history_plot.addItem(score_history_curve_list[n][i])
181 cycle_index += 1
182 self.score_history_curves_close = score_history_curve_list[0]
183 self.score_history_curves_far = score_history_curve_list[1]
184
185 def update(self, processor_result: ProcessorResult):
186 def is_none_or_detection(x):
187 return x.detection if x is not None else None
188
189 detection = np.array(
190 [
191 is_none_or_detection(processor_result.close),
192 is_none_or_detection(processor_result.far),
193 ]
194 )
195 self.detection_history = np.roll(self.detection_history, -1, axis=1)
196 self.detection_history[:, -1] = detection
197
198 self.detection_history_curve_close.setData(self.detection_history[0])
199 self.detection_history_curve_far.setData(self.detection_history[1])
200
201 if processor_result.close is not None:
202 if processor_result.close.detection:
203 self.close_text_item.show()
204 else:
205 self.close_text_item.hide()
206
207 if processor_result.far is not None:
208 if processor_result.far.detection:
209 self.far_text_item.show()
210 else:
211 self.far_text_item.hide()
212
213 max_val = 0.0
214
215 def is_none_or_threshold(x):
216 return x.threshold if x is not None else None
217
218 threshold = np.array(
219 [
220 is_none_or_threshold(processor_result.close),
221 is_none_or_threshold(processor_result.far),
222 ]
223 )
224 if np.nanmax(np.array(threshold, dtype=float)) > max_val:
225 max_val = np.nanmax(np.array(threshold, dtype=float))
226 self.threshold_history = np.roll(self.threshold_history, -1, axis=1)
227 self.threshold_history[:, -1] = threshold
228
229 self.threshold_history_curve_close.setData(self.threshold_history[0])
230 self.threshold_history_curve_far.setData(self.threshold_history[1])
231
232 if self.score_history_close is not None:
233 self.score_history_close = np.roll(self.score_history_close, -1, axis=1)
234 assert processor_result.close is not None
235 # Plot the second highest score
236 self.score_history_close[:, -1] = np.sort(processor_result.close.score, axis=0)[-2, :]
237
238 assert self.score_history_curves_close is not None
239 for i, curve in enumerate(self.score_history_curves_close):
240 # Assign x-values so that setData() doesn't give error when y-values are NaN
241 curve.setData(np.arange(0, 100), self.score_history_close[i, :].flatten())
242
243 if np.max(processor_result.close.score) > max_val:
244 max_val = np.max(processor_result.close.score)
245
246 if self.score_history_far is not None:
247 self.score_history_far = np.roll(self.score_history_far, -1, axis=1)
248 assert processor_result.far is not None
249 # Plot the second highest score
250 self.score_history_far[:, -1] = np.sort(processor_result.far.score, axis=0)[-2, :]
251
252 assert self.score_history_curves_far is not None
253 for i, curve in enumerate(self.score_history_curves_far):
254 # Assign x-values so that setData() doesn't give error when y-values are NaN
255 curve.setData(np.arange(0, 100), self.score_history_far[i, :].flatten())
256
257 if np.max(processor_result.far.score) > max_val:
258 max_val = np.max(processor_result.far.score)
259
260 if max_val != 0.0:
261 self.score_history_plot.setYRange(0.0, self.score_smooth_max.update(max_val))
262
263 @staticmethod
264 def _create_detection_plot(parent: pg.GraphicsLayout) -> pg.PlotItem:
265 detection_history_plot = parent.addPlot(row=0, col=0)
266 detection_history_plot.setTitle("Detection")
267 detection_history_plot.setLabel(axis="bottom", text="Frames")
268 detection_history_plot.setMenuEnabled(False)
269 detection_history_plot.setMouseEnabled(x=False, y=False)
270 detection_history_plot.hideButtons()
271 detection_history_plot.showGrid(x=True, y=True, alpha=0.5)
272 detection_history_plot.setYRange(-0.1, 1.8)
273 return detection_history_plot
274
275 @staticmethod
276 def _create_score_plot(parent: pg.GraphicsLayout) -> pg.PlotItem:
277 score_history_plot = parent.addPlot(row=1, col=0)
278 score_history_plot.setTitle("Detection score")
279 score_history_plot.setLabel(axis="bottom", text="Frames")
280 score_history_plot.addLegend()
281 score_history_plot.setMenuEnabled(False)
282 score_history_plot.setMouseEnabled(x=False, y=False)
283 score_history_plot.hideButtons()
284 score_history_plot.showGrid(x=True, y=True, alpha=0.5)
285 return score_history_plot
286
287
288if __name__ == "__main__":
289 main()
View this example on GitHub: acconeer/acconeer-python-exploration