examples/a121/algo/presence/processor.py#
1# Copyright (c) Acconeer AB, 2022-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 import QtCore
11
12import pyqtgraph as pg
13
14import acconeer.exptool as et
15from acconeer.exptool import a121
16from acconeer.exptool.a121.algo._utils import get_distances_m
17from acconeer.exptool.a121.algo.presence import Processor, ProcessorConfig
18
19
20def main():
21 args = a121.ExampleArgumentParser().parse_args()
22 et.utils.config_logging(args)
23
24 client = a121.Client.open(**a121.get_client_args(args))
25
26 sensor_config = a121.SensorConfig(
27 start_point=600,
28 step_length=120,
29 num_points=5,
30 profile=a121.Profile.PROFILE_5,
31 sweeps_per_frame=32,
32 hwaas=32,
33 frame_rate=20,
34 )
35
36 metadata = client.setup_session(sensor_config)
37
38 presence_config = ProcessorConfig(
39 intra_detection_threshold=1.3,
40 inter_detection_threshold=1,
41 )
42
43 presence_processor = Processor(
44 sensor_config=sensor_config,
45 metadata=metadata,
46 processor_config=presence_config,
47 )
48
49 pg_updater = PGUpdater(sensor_config, presence_config, metadata)
50 pg_process = et.PGProcess(pg_updater)
51 pg_process.start()
52
53 client.start_session()
54
55 interrupt_handler = et.utils.ExampleInterruptHandler()
56
57 while not interrupt_handler.got_signal:
58 result = client.get_next()
59 processed_data = presence_processor.process(result)
60 try:
61 pg_process.put_data(processed_data)
62 except et.PGProccessDiedException:
63 break
64
65 print("Disconnecting...")
66 pg_process.close()
67 client.close()
68
69
70class PGUpdater:
71 def __init__(self, sensor_config, processor_config, metadata):
72 self.sensor_config = sensor_config
73 self.processor_config = processor_config
74
75 self.history_length_s = 5
76 self.history_length_n = int(round(self.history_length_s * sensor_config.frame_rate))
77 self.intra_history = np.zeros(self.history_length_n)
78 self.inter_history = np.zeros(self.history_length_n)
79
80 self.distances = get_distances_m(self.sensor_config, metadata)
81
82 self.setup_is_done = False
83
84 def setup(self, win):
85 win.setWindowTitle("Acconeer presence detection example")
86
87 self.intra_limit_lines = []
88 self.inter_limit_lines = []
89
90 # Noise estimation plot
91
92 self.noise_plot = win.addPlot(
93 row=0,
94 col=0,
95 title="Noise",
96 )
97 self.noise_plot.setMenuEnabled(False)
98 self.noise_plot.setMouseEnabled(x=False, y=False)
99 self.noise_plot.hideButtons()
100 self.noise_plot.showGrid(x=True, y=True)
101 self.noise_plot.setLabel("bottom", "Distance (m)")
102 self.noise_plot.setLabel("left", "Amplitude")
103 self.noise_plot.setVisible(False)
104 self.noise_curve = self.noise_plot.plot(pen=et.utils.pg_pen_cycler())
105 self.noise_smooth_max = et.utils.SmoothMax(self.sensor_config.frame_rate)
106
107 # Depthwise presence plot
108
109 self.move_plot = pg.PlotItem(title="Depthwise presence")
110 self.move_plot.setMenuEnabled(False)
111 self.move_plot.setMouseEnabled(x=False, y=False)
112 self.move_plot.hideButtons()
113 self.move_plot.showGrid(x=True, y=True)
114 self.move_plot.setLabel("bottom", "Distance (m)")
115 self.move_plot.setLabel("left", "Norm. ampl.")
116 self.move_plot.setXRange(self.distances[0], self.distances[-1])
117 self.intra_curve = self.move_plot.plot(pen=et.utils.pg_pen_cycler(1))
118 if not self.processor_config.intra_enable:
119 self.intra_curve.hide()
120
121 self.inter_curve = self.move_plot.plot(pen=et.utils.pg_pen_cycler(0))
122 if not self.processor_config.inter_enable:
123 self.inter_curve.hide()
124
125 self.move_smooth_max = et.utils.SmoothMax(
126 self.sensor_config.frame_rate,
127 tau_decay=1.0,
128 tau_grow=0.25,
129 )
130
131 self.move_depth_line = pg.InfiniteLine(pen=pg.mkPen("k", width=1.5))
132 self.move_depth_line.hide()
133 self.move_plot.addItem(self.move_depth_line)
134
135 self.present_html_format = (
136 '<div style="text-align: center">'
137 '<span style="color: #FFFFFF;font-size:15pt;">'
138 "{}</span></div>"
139 )
140 not_present_html = (
141 '<div style="text-align: center">'
142 '<span style="color: #FFFFFF;font-size:15pt;">'
143 "{}</span></div>".format("No presence detected")
144 )
145 self.present_text_item = pg.TextItem(
146 fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
147 anchor=(0.5, 0),
148 )
149 self.not_present_text_item = pg.TextItem(
150 html=not_present_html,
151 fill=pg.mkColor(0x1F, 0x77, 0xB4, 180),
152 anchor=(0.5, 0),
153 )
154
155 self.move_plot.addItem(self.present_text_item)
156 self.move_plot.addItem(self.not_present_text_item)
157 self.present_text_item.hide()
158 self.not_present_text_item.hide()
159
160 # Intra presence history plot
161
162 self.intra_hist_plot = win.addPlot(
163 row=1,
164 col=0,
165 title="Intra presence history (fast motions)",
166 )
167 self.intra_hist_plot.setMenuEnabled(False)
168 self.intra_hist_plot.setMouseEnabled(x=False, y=False)
169 self.intra_hist_plot.hideButtons()
170 self.intra_hist_plot.showGrid(x=True, y=True)
171 self.intra_hist_plot.setLabel("bottom", "Time (s)")
172 self.intra_hist_plot.setLabel("left", "Score")
173 self.intra_hist_plot.setXRange(-self.history_length_s, 0)
174 self.intra_history_smooth_max = et.utils.SmoothMax(self.sensor_config.frame_rate)
175 self.intra_hist_plot.setYRange(0, 10)
176 if not self.processor_config.intra_enable:
177 intra_color = et.utils.color_cycler(1)
178 intra_color = f"{intra_color}50"
179 intra_dashed_pen = pg.mkPen(intra_color, width=2.5, style=QtCore.Qt.DashLine)
180 intra_pen = pg.mkPen(intra_color, width=2)
181 else:
182 intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
183 intra_pen = et.utils.pg_pen_cycler(1)
184
185 self.intra_hist_curve = self.intra_hist_plot.plot(pen=intra_pen)
186 limit_line = pg.InfiniteLine(angle=0, pen=intra_dashed_pen)
187 self.intra_hist_plot.addItem(limit_line)
188 self.intra_limit_lines.append(limit_line)
189
190 for line in self.intra_limit_lines:
191 line.setPos(self.processor_config.intra_detection_threshold)
192
193 # Inter presence history plot
194
195 self.inter_hist_plot = win.addPlot(
196 row=1,
197 col=1,
198 title="Inter presence history (slow motions)",
199 )
200 self.inter_hist_plot.setMenuEnabled(False)
201 self.inter_hist_plot.setMouseEnabled(x=False, y=False)
202 self.inter_hist_plot.hideButtons()
203 self.inter_hist_plot.showGrid(x=True, y=True)
204 self.inter_hist_plot.setLabel("bottom", "Time (s)")
205 self.inter_hist_plot.setLabel("left", "Score")
206 self.inter_hist_plot.setXRange(-self.history_length_s, 0)
207 self.inter_history_smooth_max = et.utils.SmoothMax(self.sensor_config.frame_rate)
208 self.inter_hist_plot.setYRange(0, 10)
209 if not self.processor_config.inter_enable:
210 inter_color = et.utils.color_cycler(0)
211 inter_color = f"{inter_color}50"
212 inter_dashed_pen = pg.mkPen(inter_color, width=2.5, style=QtCore.Qt.DashLine)
213 inter_pen = pg.mkPen(inter_color, width=2)
214 else:
215 inter_pen = et.utils.pg_pen_cycler(0)
216 inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
217
218 self.inter_hist_curve = self.inter_hist_plot.plot(pen=inter_pen)
219 limit_line = pg.InfiniteLine(angle=0, pen=inter_dashed_pen)
220 self.inter_hist_plot.addItem(limit_line)
221 self.inter_limit_lines.append(limit_line)
222
223 for line in self.inter_limit_lines:
224 line.setPos(self.processor_config.inter_detection_threshold)
225
226 sublayout = win.addLayout(row=2, col=0, colspan=2)
227 sublayout.layout.setColumnStretchFactor(0, 2)
228 sublayout.addItem(self.move_plot, row=0, col=0)
229
230 self.setup_is_done = True
231
232 def update(self, data):
233 noise = data.extra_result.lp_noise
234 self.noise_curve.setData(self.distances, noise)
235 self.noise_plot.setYRange(0, self.noise_smooth_max.update(noise))
236
237 movement_x = data.presence_distance
238
239 self.inter_curve.setData(self.distances, data.inter)
240 self.intra_curve.setData(self.distances, data.intra)
241 m = self.move_smooth_max.update(np.max(np.maximum(data.inter, data.intra)))
242 m = max(
243 m,
244 2
245 * np.maximum(
246 self.processor_config.intra_detection_threshold,
247 self.processor_config.inter_detection_threshold,
248 ),
249 )
250 self.move_plot.setYRange(0, m)
251 self.move_depth_line.setPos(movement_x)
252 self.move_depth_line.setVisible(bool(data.presence_detected))
253
254 self.set_present_text_y_pos(m)
255
256 if data.presence_detected:
257 present_text = f"Presence detected at {movement_x * 100:.0f} cm"
258 present_html = self.present_html_format.format(present_text)
259 self.present_text_item.setHtml(present_html)
260
261 self.present_text_item.show()
262 self.not_present_text_item.hide()
263 else:
264 self.present_text_item.hide()
265 self.not_present_text_item.show()
266
267 # Intra presence
268
269 move_hist_xs = np.linspace(-self.history_length_s, 0, self.history_length_n)
270
271 self.intra_history = np.roll(self.intra_history, -1)
272 self.intra_history[-1] = data.intra_presence_score
273
274 m_hist = max(
275 float(np.max(self.intra_history)),
276 self.processor_config.intra_detection_threshold * 1.05,
277 )
278 m_hist = self.intra_history_smooth_max.update(m_hist)
279
280 self.intra_hist_plot.setYRange(0, m_hist)
281 self.intra_hist_curve.setData(move_hist_xs, self.intra_history)
282
283 # Inter presence
284
285 self.inter_history = np.roll(self.inter_history, -1)
286 self.inter_history[-1] = data.inter_presence_score
287
288 m_hist = max(
289 float(np.max(self.inter_history)),
290 self.processor_config.inter_detection_threshold * 1.05,
291 )
292 m_hist = self.inter_history_smooth_max.update(m_hist)
293
294 self.inter_hist_plot.setYRange(0, m_hist)
295 self.inter_hist_curve.setData(move_hist_xs, self.inter_history)
296
297 def set_present_text_y_pos(self, y):
298 x_pos = self.distances[0] + (self.distances[-1] - self.distances[0]) / 2
299 self.present_text_item.setPos(x_pos, 0.95 * y)
300 self.not_present_text_item.setPos(x_pos, 0.95 * y)
301
302
303if __name__ == "__main__":
304 main()
View this example on GitHub: acconeer/acconeer-python-exploration