examples/a121/algo/cargo/example_app.py#
1# Copyright (c) Acconeer AB, 2025-2026
2# All rights reserved
3
4from __future__ import annotations
5
6import numpy as np
7
8# Added here to force pyqtgraph to choose PySide
9import PySide6 # noqa: F401
10from PySide6.QtGui import QFont
11
12import pyqtgraph as pg
13
14import acconeer.exptool as et
15from acconeer.exptool import a121
16from acconeer.exptool.a121.algo.cargo import (
17 CargoPresenceConfig,
18 ExApp,
19 ExAppConfig,
20 ExAppContext,
21 ExAppResult,
22 UtilizationLevelConfig,
23)
24
25# These are some of the presets available.
26# Replace 'ex_app_config' further down to use one of the presets
27from acconeer.exptool.a121.algo.cargo._configs import (
28 get_10_ft_container_config,
29 get_20_ft_container_config,
30 get_40_ft_container_config,
31)
32from acconeer.exptool.a121.algo.cargo._ex_app import PRESENCE_RUN_TIME_S, ContainerSize, _Mode
33
34
35SENSOR_ID = 1
36
37
38def main():
39 args = a121.ExampleArgumentParser().parse_args()
40 et.utils.config_logging(args)
41
42 client = a121.Client.open(**a121.get_client_args(args))
43
44 # All configurable parameters are displayed here.
45 # These settings correspond to the "No lens" preset.
46 # There are also other preset configurations available by the imports!
47
48 utilization_level_config = UtilizationLevelConfig(
49 update_rate=5,
50 threshold_sensitivity=0.5,
51 signal_quality=25,
52 )
53
54 cargo_presence_config = CargoPresenceConfig(
55 burst_rate=0.1,
56 update_rate=6,
57 signal_quality=30,
58 sweeps_per_frame=12,
59 inter_detection_threshold=2,
60 intra_detection_threshold=2,
61 )
62
63 ex_app_config = ExAppConfig(
64 activate_presence=True,
65 cargo_presence_config=cargo_presence_config,
66 activate_utilization_level=True,
67 utilization_level_config=utilization_level_config,
68 container_size=ContainerSize.CONTAINER_20_FT,
69 )
70
71 example_app = ExApp(
72 client=client,
73 sensor_id=SENSOR_ID,
74 ex_app_config=ex_app_config,
75 )
76 example_app.start()
77
78 pg_updater = PGUpdater(
79 ex_app_config,
80 example_app.ex_app_context,
81 )
82
83 pg_process = et.PGProcess(pg_updater)
84 pg_process.start()
85
86 interrupt_handler = et.utils.ExampleInterruptHandler()
87
88 while not interrupt_handler.got_signal:
89 example_app_result = example_app.get_next()
90 if example_app_result.mode == _Mode.DISTANCE:
91 if example_app_result.level_m is not None:
92 result_str = (
93 f"Utilization level: {np.around(example_app_result.level_m, 2)} m, "
94 f"{int(np.around(example_app_result.level_percent, 0))}%"
95 )
96 else:
97 result_str = "No utilization level detected"
98 else:
99 if example_app_result.presence_detected:
100 result_str = "Presence detected"
101 else:
102 result_str = "Presence not detected"
103 print(f"Running detector: {example_app_result.mode.name}\n{result_str}\n")
104
105 try:
106 pg_process.put_data(example_app_result)
107 except et.PGProccessDiedException:
108 break
109
110 print("Disconnecting...")
111 pg_process.close()
112 client.close()
113
114
115class PGUpdater:
116 def __init__(self, ex_app_config: ExAppConfig, ex_app_context: ExAppContext):
117 self.ex_app_config = ex_app_config
118 self.ex_app_context = ex_app_context
119
120 def setup(self, win) -> None:
121 self.utilization_level_config = self.ex_app_config.utilization_level_config
122 self.cargo_presence_config = self.ex_app_config.cargo_presence_config
123
124 self.num_rects = 16
125
126 # Utilization level
127
128 if self.ex_app_config.activate_utilization_level:
129 # Distance sweep plot
130
131 self.sweep_plot = win.addPlot(row=0, col=0, title="Distance sweep")
132 self.sweep_plot.setMenuEnabled(False)
133 self.sweep_plot.showGrid(x=True, y=True)
134 self.sweep_plot.addLegend()
135 self.sweep_plot.setLabel("left", "Amplitude")
136 self.sweep_plot.setLabel("bottom", "Distance (m)")
137 self.sweep_plot.addItem(pg.PlotDataItem())
138
139 self.num_curves = 4
140
141 pen = et.utils.pg_pen_cycler(0)
142 brush = et.utils.pg_brush_cycler(0)
143 symbol_kw = dict(symbol="o", symbolSize=1, symbolBrush=brush, symbolPen="k")
144 feat_kw = dict(pen=pen, **symbol_kw)
145 self.sweep_curves = [self.sweep_plot.plot(**feat_kw) for _ in range(self.num_curves)]
146
147 pen = et.utils.pg_pen_cycler(1)
148 brush = et.utils.pg_brush_cycler(1)
149 symbol_kw = dict(symbol="o", symbolSize=1, symbolBrush=brush, symbolPen="k")
150 feat_kw = dict(pen=pen, **symbol_kw)
151 self.threshold_curves = [
152 self.sweep_plot.plot(**feat_kw) for _ in range(self.num_curves)
153 ]
154
155 sweep_plot_legend = pg.LegendItem(offset=(-30, 30))
156 sweep_plot_legend.setParentItem(self.sweep_plot.graphicsItem())
157 sweep_plot_legend.addItem(self.sweep_curves[0], "Sweep")
158 sweep_plot_legend.addItem(self.threshold_curves[0], "Threshold")
159
160 font = QFont()
161 font.setPixelSize(16)
162 self.sweep_text_item = pg.TextItem(
163 fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
164 anchor=(0.5, 0),
165 color=pg.mkColor(0xFF, 0xFF, 0xFF, 200),
166 )
167 self.sweep_text_item.setFont(font)
168 self.sweep_text_item.hide()
169 self.sweep_plot.addItem(self.sweep_text_item)
170
171 self.sweep_main_peak_line = pg.InfiniteLine(pen=pg.mkPen("k", width=1.5, dash=[2, 8]))
172 self.sweep_main_peak_line.hide()
173 self.sweep_plot.addItem(self.sweep_main_peak_line)
174
175 self.sweep_smooth_max = et.utils.SmoothMax()
176
177 # Utilization level plot
178
179 self.rect_plot = win.addPlot(row=1, col=0, title="Utilization level")
180 self.rect_plot.setAspectLocked()
181 self.rect_plot.hideAxis("left")
182 self.rect_plot.hideAxis("bottom")
183 self.rects = []
184
185 pen = pg.mkPen(None)
186 rect_height = self.num_rects / 2.0
187 for r in np.arange(self.num_rects) + 1:
188 rect = pg.QtWidgets.QGraphicsRectItem(r, rect_height, 1, rect_height)
189 rect.setPen(pen)
190 rect.setBrush(et.utils.pg_brush_cycler(7))
191 self.rect_plot.addItem(rect)
192 self.rects.append(rect)
193
194 self.level_html_format = (
195 '<div style="text-align: center">'
196 '<span style="color: #FFFFFF;font-size:12pt;">'
197 "{}</span></div>"
198 )
199
200 self.level_text_item = pg.TextItem(
201 fill=pg.mkColor(0, 150, 0),
202 anchor=(0.5, 0),
203 )
204
205 no_detection_html = (
206 '<div style="text-align: center">'
207 '<span style="color: #FFFFFF;font-size:12pt;">'
208 "{}</span></div>".format("No level detected")
209 )
210
211 self.no_detection_text_item = pg.TextItem(
212 html=no_detection_html,
213 fill=pg.mkColor("r"),
214 anchor=(0.5, 0),
215 )
216
217 self.rect_plot.addItem(self.level_text_item)
218 self.rect_plot.addItem(self.no_detection_text_item)
219 self.level_text_item.setPos(self.num_rects / 2.0 + 1, self.num_rects + 2)
220 self.level_text_item.hide()
221 self.no_detection_text_item.setPos(self.num_rects / 2.0 + 1, self.num_rects + 2)
222 self.no_detection_text_item.show()
223
224 # Presence
225
226 if self.ex_app_config.activate_presence:
227 estimated_frame_rate = self.ex_app_context.presence_context.estimated_frame_rate
228
229 self.history_length_n = int(round(PRESENCE_RUN_TIME_S * estimated_frame_rate) + 1)
230 self.intra_history = np.zeros(self.history_length_n)
231 self.inter_history = np.zeros(self.history_length_n)
232
233 # Presence history plot
234
235 self.presence_hist_plot = win.addPlot(
236 row=0,
237 col=1,
238 title="Presence history",
239 )
240 self.presence_hist_plot.setMenuEnabled(False)
241 self.presence_hist_plot.setMouseEnabled(x=False, y=False)
242 self.presence_hist_plot.hideButtons()
243 self.presence_hist_plot.showGrid(x=True, y=True)
244 self.presence_hist_plot.setLabel("bottom", "Time (s)")
245 self.presence_hist_plot.setLabel("left", "Score")
246 self.presence_hist_plot.setXRange(-PRESENCE_RUN_TIME_S, 0)
247 self.presence_history_smooth_max = et.utils.SmoothMax(estimated_frame_rate)
248 self.presence_hist_plot.setYRange(0, 10)
249
250 self.intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
251 self.intra_pen = et.utils.pg_pen_cycler(1)
252
253 self.intra_hist_curve = self.presence_hist_plot.plot(pen=self.intra_pen)
254 self.intra_limit_line = pg.InfiniteLine(angle=0, pen=self.intra_dashed_pen)
255 self.presence_hist_plot.addItem(self.intra_limit_line)
256 self.intra_limit_line.setPos(self.cargo_presence_config.intra_detection_threshold)
257 self.intra_limit_line.setPen(self.intra_dashed_pen)
258
259 self.inter_pen = et.utils.pg_pen_cycler(0)
260 self.inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
261
262 self.inter_hist_curve = self.presence_hist_plot.plot(pen=self.inter_pen)
263 self.inter_limit_line = pg.InfiniteLine(angle=0, pen=self.inter_dashed_pen)
264 self.presence_hist_plot.addItem(self.inter_limit_line)
265 self.inter_limit_line.setPos(self.cargo_presence_config.inter_detection_threshold)
266 self.inter_limit_line.setPen(self.inter_dashed_pen)
267
268 self.hist_xs = np.linspace(-PRESENCE_RUN_TIME_S, 0, self.history_length_n)
269
270 # Presence detection plot
271
272 self.presence_detection_plot = win.addPlot(row=1, col=1, title="Presence")
273 self.presence_detection_plot.setAspectLocked()
274 self.presence_detection_plot.hideAxis("left")
275 self.presence_detection_plot.hideAxis("bottom")
276
277 present_html_format = (
278 '<div style="text-align: center">'
279 '<span style="color: #FFFFFF;font-size:12pt;">'
280 "{}</span></div>".format("Presence detected")
281 )
282 not_present_html = (
283 '<div style="text-align: center">'
284 '<span style="color: #FFFFFF;font-size:12pt;">'
285 "{}</span></div>".format("No presence detected")
286 )
287 self.present_text_item = pg.TextItem(
288 html=present_html_format,
289 fill=pg.mkColor(0, 150, 0),
290 anchor=(0.65, 0),
291 )
292 self.not_present_text_item = pg.TextItem(
293 html=not_present_html,
294 fill=pg.mkColor("r"),
295 anchor=(0.6, 0),
296 )
297
298 self.presence_detection_plot.addItem(self.present_text_item)
299 self.presence_detection_plot.addItem(self.not_present_text_item)
300 self.present_text_item.setPos(self.num_rects / 2.0 + 1, self.num_rects + 2)
301 self.not_present_text_item.setPos(self.num_rects / 2.0 + 1, self.num_rects + 2)
302 self.present_text_item.hide()
303
304 pen = pg.mkPen(None)
305 rect_height = self.num_rects / 2.0
306 rect_length = self.num_rects
307
308 self.presence_rect = pg.QtWidgets.QGraphicsRectItem(
309 0, rect_height, rect_length, rect_height
310 )
311 self.presence_rect.setPen(pen)
312 self.presence_rect.setBrush(et.utils.pg_brush_cycler(7))
313 self.presence_detection_plot.addItem(self.presence_rect)
314
315 def update(self, result: ExAppResult) -> None:
316 if result.mode == _Mode.DISTANCE:
317 if self.ex_app_config.activate_presence:
318 self.intra_history = np.zeros(self.history_length_n)
319 self.inter_history = np.zeros(self.history_length_n)
320
321 # Sweep plot
322
323 distance = result.distance
324 max_val_in_plot = 0
325 if result.distance_processor_result is not None:
326 for idx, processor_result in enumerate(result.distance_processor_result):
327 abs_sweep = processor_result.extra_result.abs_sweep
328 threshold = processor_result.extra_result.used_threshold
329 distances_m = processor_result.extra_result.distances_m
330
331 self.sweep_curves[idx].setData(distances_m, abs_sweep)
332 self.threshold_curves[idx].setData(distances_m, threshold)
333
334 max_val_in_subsweep = max(max(threshold), max(abs_sweep))
335 if max_val_in_plot < max_val_in_subsweep:
336 max_val_in_plot = max_val_in_subsweep
337
338 self.sweep_plot.setYRange(0, self.sweep_smooth_max.update(max_val_in_plot))
339
340 if distance is not None:
341 text_y_pos = self.sweep_plot.getAxis("left").range[1] * 0.95
342 text_x_pos = (
343 self.sweep_plot.getAxis("bottom").range[1]
344 + self.sweep_plot.getAxis("bottom").range[0]
345 ) / 2.0
346 self.sweep_text_item.setPos(text_x_pos, text_y_pos)
347 self.sweep_text_item.setHtml(f"Main peak distance: {distance:.3f} m")
348 self.sweep_text_item.show()
349
350 self.sweep_main_peak_line.setPos(distance)
351 self.sweep_main_peak_line.show()
352 else:
353 self.sweep_text_item.hide()
354 self.sweep_main_peak_line.hide()
355
356 # Utilization level plot
357
358 # Show the percentage level plot if the plot width is greater than 400 pixels,
359 # otherwise display the level as text.
360
361 if result.level_percent is None: # No detection
362 for rect in self.rects:
363 rect.setBrush(et.utils.pg_brush_cycler(7))
364 self.level_text_item.hide()
365 self.no_detection_text_item.show()
366 else:
367 self.bar_loc = int(
368 np.around(self.num_rects - result.level_percent / 100 * self.num_rects)
369 )
370 for rect in self.rects[self.bar_loc :]:
371 rect.setBrush(et.utils.pg_brush_cycler(0))
372
373 for rect in self.rects[: self.bar_loc]:
374 rect.setBrush(et.utils.pg_brush_cycler(7))
375
376 level_text = (
377 f"Utilization level: {result.level_m:.2f} m, {result.level_percent:.0f} %"
378 )
379 level_html = self.level_html_format.format(level_text)
380 self.level_text_item.setHtml(level_html)
381 self.level_text_item.show()
382 self.no_detection_text_item.hide()
383
384 for rect in self.rects:
385 rect.setVisible(True)
386
387 else:
388 # Presence history
389
390 self.intra_history = np.roll(self.intra_history, -1)
391 self.intra_history[-1] = result.intra_presence_score
392 self.intra_hist_curve.setData(self.hist_xs, self.intra_history)
393
394 self.inter_history = np.roll(self.inter_history, -1)
395 self.inter_history[-1] = result.inter_presence_score
396 self.inter_hist_curve.setData(self.hist_xs, self.inter_history)
397
398 # Set y-range
399
400 if np.isnan(self.intra_history).all():
401 intra_m_hist = self.cargo_presence_config.intra_detection_threshold
402 else:
403 intra_m_hist = max(
404 float(np.nanmax(self.intra_history)),
405 self.cargo_presence_config.intra_detection_threshold * 1.05,
406 )
407
408 if np.isnan(self.inter_history).all():
409 inter_m_hist = self.cargo_presence_config.inter_detection_threshold
410 else:
411 inter_m_hist = max(
412 float(np.nanmax(self.inter_history)),
413 self.cargo_presence_config.inter_detection_threshold * 1.05,
414 )
415
416 m_hist = max(intra_m_hist, inter_m_hist)
417 m_hist = self.presence_history_smooth_max.update(m_hist)
418 self.presence_hist_plot.setYRange(0, m_hist)
419
420 # Presence detection plot
421
422 if result.presence_detected:
423 self.present_text_item.show()
424 self.not_present_text_item.hide()
425 self.presence_rect.setBrush(pg.mkColor(0, 150, 0))
426 else:
427 self.present_text_item.hide()
428 self.not_present_text_item.show()
429 self.presence_rect.setBrush(et.utils.pg_brush_cycler(7))
430
431
432if __name__ == "__main__":
433 main()
View this example on GitHub: acconeer/acconeer-python-exploration