examples/a121/algo/smart_presence/ref_app.py#
1# Copyright (c) Acconeer AB, 2023-2026
2# All rights reserved
3
4from __future__ import annotations
5
6from typing import List, Tuple
7
8import numpy as np
9import numpy.typing as npt
10
11from PySide6 import QtCore
12
13import pyqtgraph as pg
14
15import acconeer.exptool as et
16from acconeer.exptool import a121
17from acconeer.exptool.a121.algo.smart_presence._ref_app import (
18 PresenceWakeUpConfig,
19 PresenceZoneConfig,
20 RefApp,
21 RefAppConfig,
22 RefAppResult,
23 _Mode,
24)
25
26
27def main():
28 args = a121.ExampleArgumentParser().parse_args()
29 et.utils.config_logging(args)
30
31 client = a121.Client.open(**a121.get_client_args(args))
32
33 ref_app_config = RefAppConfig(
34 wake_up_mode=True,
35 wake_up_config=PresenceWakeUpConfig(
36 start_m=1.0,
37 end_m=3.0,
38 num_zones=5,
39 num_zones_for_wake_up=2,
40 ),
41 nominal_config=PresenceZoneConfig(
42 start_m=1.0,
43 end_m=3.0,
44 num_zones=3,
45 ),
46 )
47
48 ref_app = RefApp(client=client, sensor_id=1, ref_app_config=ref_app_config)
49 ref_app.start()
50
51 nominal_sensor_config = ref_app.nominal_detector_config.to_sensor_config()
52 distances = np.linspace(
53 ref_app_config.nominal_config.start_m,
54 ref_app_config.nominal_config.end_m,
55 nominal_sensor_config.num_points,
56 )
57 nominal_zone_limits = ref_app.ref_app_processor.create_zones(
58 distances, ref_app_config.nominal_config.num_zones
59 )
60
61 pg_updater = PGUpdater(
62 ref_app_config,
63 ref_app.ref_app_context.wake_up_detector_context.estimated_frame_rate,
64 nominal_zone_limits,
65 ref_app.ref_app_processor.zone_limits,
66 )
67 pg_process = et.PGProcess(pg_updater)
68 pg_process.start()
69
70 interrupt_handler = et.utils.ExampleInterruptHandler()
71 print("Press Ctrl-C to end session")
72
73 while not interrupt_handler.got_signal:
74 ref_app_result = ref_app.get_next()
75 if ref_app_result.presence_detected:
76 print(f"Presence in zone {ref_app_result.max_presence_zone}")
77 else:
78 print("No presence")
79 try:
80 pg_process.put_data(ref_app_result)
81 except et.PGProccessDiedException:
82 break
83
84 ref_app.stop()
85
86 print("Disconnecting...")
87 client.close()
88
89
90class PGUpdater:
91 def __init__(
92 self,
93 ref_app_config: RefAppConfig,
94 estimated_frame_rate: float,
95 nominal_zone_limits: npt.NDArray[np.float64],
96 wake_up_zone_limits: npt.NDArray[np.float64],
97 ) -> None:
98 self.ref_app_config = ref_app_config
99 self.nominal_config = ref_app_config.nominal_config
100 self.wake_up_config = ref_app_config.wake_up_config
101
102 self.show_all_detected_zones = ref_app_config.show_all_detected_zones
103 self.nominal_zone_limits = nominal_zone_limits
104 self.wake_up_zone_limits = wake_up_zone_limits
105 self.estimated_frame_rate = estimated_frame_rate
106
107 self.history_length_s = 5
108 self.time_fifo: List[float] = []
109 self.intra_fifo: List[float] = []
110 self.inter_fifo: List[float] = []
111
112 self.intra_limit_lines = []
113 self.inter_limit_lines = []
114
115 self.setup_is_done = False
116
117 def setup(self, win):
118 win.setWindowTitle("Acconeer smart presence example")
119
120 # Intra presence history plot
121
122 self.intra_hist_plot = win.addPlot(
123 row=0,
124 col=0,
125 title="Intra presence history (fast motions)",
126 )
127 self.intra_hist_plot.setMenuEnabled(False)
128 self.intra_hist_plot.setMouseEnabled(x=False, y=False)
129 self.intra_hist_plot.hideButtons()
130 self.intra_hist_plot.showGrid(x=True, y=True)
131 self.intra_hist_plot.setLabel("bottom", "Time (s)")
132 self.intra_hist_plot.setLabel("left", "Score")
133 self.intra_hist_plot.setXRange(-self.history_length_s, 0)
134 self.intra_history_smooth_max = et.utils.SmoothMax(self.estimated_frame_rate)
135 self.intra_hist_plot.setYRange(0, 10)
136 if not self.nominal_config.intra_enable:
137 intra_color = et.utils.color_cycler(1)
138 intra_color = f"{intra_color}50"
139 self.nominal_intra_dashed_pen = pg.mkPen(
140 intra_color, width=2.5, style=QtCore.Qt.DashLine
141 )
142 self.nominal_intra_pen = pg.mkPen(intra_color, width=2)
143 else:
144 self.nominal_intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
145 self.nominal_intra_pen = et.utils.pg_pen_cycler(1)
146
147 self.intra_hist_curve = self.intra_hist_plot.plot(pen=self.nominal_intra_pen)
148 limit_line = pg.InfiniteLine(angle=0, pen=self.nominal_intra_dashed_pen)
149 self.intra_hist_plot.addItem(limit_line)
150 self.intra_limit_lines.append(limit_line)
151
152 for line in self.intra_limit_lines:
153 line.setPos(self.nominal_config.intra_detection_threshold)
154
155 # Inter presence history plot
156
157 self.inter_hist_plot = win.addPlot(
158 row=0,
159 col=1,
160 title="Inter presence history (slow motions)",
161 )
162 self.inter_hist_plot.setMenuEnabled(False)
163 self.inter_hist_plot.setMouseEnabled(x=False, y=False)
164 self.inter_hist_plot.hideButtons()
165 self.inter_hist_plot.showGrid(x=True, y=True)
166 self.inter_hist_plot.setLabel("bottom", "Time (s)")
167 self.inter_hist_plot.setLabel("left", "Score")
168 self.inter_hist_plot.setXRange(-self.history_length_s, 0)
169 self.inter_history_smooth_max = et.utils.SmoothMax(self.estimated_frame_rate)
170 self.inter_hist_plot.setYRange(0, 10)
171 if not self.nominal_config.inter_enable:
172 inter_color = et.utils.color_cycler(0)
173 inter_color = f"{inter_color}50"
174 self.nominal_inter_dashed_pen = pg.mkPen(
175 inter_color, width=2.5, style=QtCore.Qt.DashLine
176 )
177 self.nominal_inter_pen = pg.mkPen(inter_color, width=2)
178 else:
179 self.nominal_inter_pen = et.utils.pg_pen_cycler(0)
180 self.nominal_inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
181
182 self.inter_hist_curve = self.inter_hist_plot.plot(pen=self.nominal_inter_pen)
183 limit_line = pg.InfiniteLine(angle=0, pen=self.nominal_inter_dashed_pen)
184 self.inter_hist_plot.addItem(limit_line)
185 self.inter_limit_lines.append(limit_line)
186
187 for line in self.inter_limit_lines:
188 line.setPos(self.nominal_config.inter_detection_threshold)
189
190 # Sector plot
191
192 if self.ref_app_config.wake_up_mode:
193 title = (
194 "Nominal config<br>"
195 "Detection type: fast (orange), slow (blue), both (green)<br>"
196 "Green background indicates active"
197 )
198 else:
199 title = "Nominal config<br>" "Detection type: fast (orange), slow (blue), both (green)"
200
201 self.nominal_sector_plot, self.nominal_sectors = self.create_sector_plot(
202 title,
203 self.ref_app_config.nominal_config.num_zones,
204 self.nominal_config.start_m,
205 self.nominal_zone_limits,
206 )
207
208 if not self.ref_app_config.wake_up_mode:
209 sublayout = win.addLayout(row=1, col=0, colspan=2)
210 sublayout.layout.setColumnStretchFactor(0, 2)
211 sublayout.addItem(self.nominal_sector_plot, row=0, col=0)
212 else:
213 assert self.wake_up_config is not None
214 sublayout = win.addLayout(row=1, col=0, colspan=2)
215 sublayout.addItem(self.nominal_sector_plot, row=0, col=1)
216
217 title = (
218 "Wake up config<br>"
219 "Detection type: fast (orange), slow (blue), both (green),<br>"
220 "lingering (light grey)<br>"
221 "Green background indicates active"
222 )
223 self.wake_up_sector_plot, self.wake_up_sectors = self.create_sector_plot(
224 title,
225 self.wake_up_config.num_zones,
226 self.wake_up_config.start_m,
227 self.wake_up_zone_limits,
228 )
229
230 sublayout.addItem(self.wake_up_sector_plot, row=0, col=0)
231
232 if self.wake_up_config.intra_enable:
233 self.wake_up_intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
234 self.wake_up_intra_pen = et.utils.pg_pen_cycler(1)
235 else:
236 intra_color = et.utils.color_cycler(1)
237 intra_color = f"{intra_color}50"
238 self.wake_up_intra_dashed_pen = pg.mkPen(
239 intra_color, width=2.5, style=QtCore.Qt.DashLine
240 )
241 self.wake_up_intra_pen = pg.mkPen(intra_color, width=2)
242
243 if self.wake_up_config.inter_enable:
244 self.wake_up_inter_pen = et.utils.pg_pen_cycler(0)
245 self.wake_up_inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
246 else:
247 inter_color = et.utils.color_cycler(0)
248 inter_color = f"{inter_color}50"
249 self.wake_up_inter_dashed_pen = pg.mkPen(
250 inter_color, width=2.5, style=QtCore.Qt.DashLine
251 )
252 self.wake_up_inter_pen = pg.mkPen(inter_color, width=2)
253
254 @staticmethod
255 def create_sector_plot(
256 title: str, num_sectors: int, start_m: float, zone_limits: npt.NDArray[np.float64]
257 ) -> Tuple[pg.PlotItem, List[pg.QtWidgets.QGraphicsEllipseItem]]:
258 sector_plot = pg.PlotItem(title=title)
259
260 sector_plot.setAspectLocked()
261 sector_plot.hideAxis("left")
262 sector_plot.hideAxis("bottom")
263
264 sectors = []
265 limit_text = []
266
267 range_html = (
268 '<div style="text-align: center">'
269 '<span style="color: #000000;font-size:12pt;">'
270 "{}</span></div>"
271 )
272
273 if start_m == zone_limits[0]:
274 x_offset = 0.7
275 else:
276 x_offset = 0
277
278 pen = pg.mkPen("k", width=1)
279 span_deg = 25
280 for r in np.flip(np.arange(1, num_sectors + 2)):
281 sector = pg.QtWidgets.QGraphicsEllipseItem(-r, -r, r * 2, r * 2)
282 sector.setStartAngle(-16 * span_deg)
283 sector.setSpanAngle(16 * span_deg * 2)
284 sector.setPen(pen)
285 sector_plot.addItem(sector)
286 sectors.append(sector)
287
288 if r != 1:
289 limit = pg.TextItem(html=range_html, anchor=(0.5, 0.5), angle=25)
290 x = r * np.cos(np.radians(span_deg))
291 y = r * np.sin(np.radians(span_deg))
292 limit.setPos(x - x_offset, y + 0.25)
293 sector_plot.addItem(limit)
294 limit_text.append(limit)
295
296 sectors.reverse()
297
298 if not start_m == zone_limits[0]:
299 start_limit_text = pg.TextItem(html=range_html, anchor=(0.5, 0.5), angle=25)
300 start_range_html = range_html.format(f"{start_m}")
301 start_limit_text.setHtml(start_range_html)
302 x = 1 * np.cos(np.radians(span_deg))
303 y = 1 * np.sin(np.radians(span_deg))
304
305 start_limit_text.setPos(x, y + 0.25)
306 sector_plot.addItem(start_limit_text)
307
308 unit_text = pg.TextItem(html=range_html, anchor=(0.5, 0.5))
309 unit_html = range_html.format("[m]")
310 unit_text.setHtml(unit_html)
311 x = (num_sectors + 2) * np.cos(np.radians(span_deg))
312 y = (num_sectors + 2) * np.sin(np.radians(span_deg))
313 unit_text.setPos(x - x_offset, y + 0.25)
314 sector_plot.addItem(unit_text)
315
316 for text_item, limit in zip(limit_text, np.flip(zone_limits)):
317 zone_range_html = range_html.format(np.around(limit, 1))
318 text_item.setHtml(zone_range_html)
319
320 return sector_plot, sectors
321
322 def update(self, data: RefAppResult) -> None:
323 if data.used_config == _Mode.NOMINAL_CONFIG:
324 inter_threshold = self.nominal_config.inter_detection_threshold
325 intra_threshold = self.nominal_config.intra_detection_threshold
326 intra_pen = self.nominal_intra_pen
327 intra_dashed_pen = self.nominal_intra_dashed_pen
328 inter_pen = self.nominal_inter_pen
329 inter_dashed_pen = self.nominal_inter_dashed_pen
330 else:
331 assert self.wake_up_config is not None
332 inter_threshold = self.wake_up_config.inter_detection_threshold
333 intra_threshold = self.wake_up_config.intra_detection_threshold
334 intra_pen = self.wake_up_intra_pen
335 intra_dashed_pen = self.wake_up_intra_dashed_pen
336 inter_pen = self.wake_up_inter_pen
337 inter_dashed_pen = self.wake_up_inter_dashed_pen
338
339 self.time_fifo.append(data.service_result.tick_time)
340
341 if data.switch_delay:
342 self.intra_fifo.append(float("nan"))
343 self.inter_fifo.append(float("nan"))
344 else:
345 self.intra_fifo.append(data.intra_presence_score)
346 self.inter_fifo.append(data.inter_presence_score)
347
348 while self.time_fifo[-1] - self.time_fifo[0] > self.history_length_s:
349 self.time_fifo.pop(0)
350 self.intra_fifo.pop(0)
351 self.inter_fifo.pop(0)
352
353 times = [t - self.time_fifo[-1] for t in self.time_fifo]
354
355 # Intra presence
356
357 if np.isnan(self.intra_fifo).all():
358 m_hist = intra_threshold
359 else:
360 m_hist = np.maximum(float(np.nanmax(self.intra_fifo)), intra_threshold * 1.05)
361
362 m_hist = self.intra_history_smooth_max.update(m_hist)
363
364 self.intra_hist_plot.setYRange(0, m_hist)
365 self.intra_hist_curve.setData(times, self.intra_fifo, connect="finite")
366 self.intra_hist_curve.setPen(intra_pen)
367
368 for line in self.intra_limit_lines:
369 line.setPos(intra_threshold)
370 line.setPen(intra_dashed_pen)
371
372 # Inter presence
373
374 if np.isnan(self.inter_fifo).all():
375 m_hist = inter_threshold
376 else:
377 m_hist = np.maximum(float(np.nanmax(self.inter_fifo)), inter_threshold * 1.05)
378
379 m_hist = self.inter_history_smooth_max.update(m_hist)
380
381 self.inter_hist_plot.setYRange(0, m_hist)
382 self.inter_hist_curve.setData(times, self.inter_fifo, connect="finite")
383 self.inter_hist_curve.setPen(inter_pen)
384
385 for line in self.inter_limit_lines:
386 line.setPos(inter_threshold)
387 line.setPen(inter_dashed_pen)
388
389 # Sector
390
391 brush = et.utils.pg_brush_cycler(7)
392 for sector in self.nominal_sectors:
393 sector.setBrush(brush)
394
395 if not self.ref_app_config.wake_up_mode:
396 sectors = self.nominal_sectors[1:]
397 show_all_zones = self.show_all_detected_zones
398 color_nominal = "white"
399 else:
400 if data.used_config == _Mode.WAKE_UP_CONFIG:
401 sectors = self.wake_up_sectors[1:]
402 show_all_zones = True
403 color_wake_up = "#DFF1D6"
404 color_nominal = "white"
405 else:
406 sectors = self.nominal_sectors[1:]
407 show_all_zones = self.show_all_detected_zones
408 color_wake_up = "white"
409 color_nominal = "#DFF1D6"
410
411 vb = self.nominal_sector_plot.getViewBox()
412 vb.setBackgroundColor(color_nominal)
413 vb = self.wake_up_sector_plot.getViewBox()
414 vb.setBackgroundColor(color_wake_up)
415
416 for sector in self.wake_up_sectors:
417 sector.setBrush(brush)
418
419 if data.presence_detected:
420 self.color_zones(data, show_all_zones, sectors)
421 self.switch_data = data
422 elif data.switch_delay:
423 self.color_zones(self.switch_data, True, self.wake_up_sectors[1:])
424
425 self.nominal_sectors[0].setPen(pg.mkPen(color_nominal, width=1))
426 self.nominal_sectors[0].setBrush(pg.mkBrush(color_nominal))
427
428 if self.ref_app_config.wake_up_mode:
429 self.wake_up_sectors[0].setPen(pg.mkPen(color_wake_up, width=1))
430 self.wake_up_sectors[0].setBrush(pg.mkBrush(color_wake_up))
431
432 @staticmethod
433 def color_zones(
434 data: RefAppResult,
435 show_all_detected_zones: bool,
436 sectors: List[pg.QtWidgets.QGraphicsEllipseItem],
437 ) -> None:
438 if show_all_detected_zones:
439 for zone, (inter_value, intra_value) in enumerate(
440 zip(data.inter_zone_detections, data.intra_zone_detections)
441 ):
442 if inter_value + intra_value == 2:
443 sectors[zone].setBrush(et.utils.pg_brush_cycler(2))
444 elif inter_value == 1:
445 sectors[zone].setBrush(et.utils.pg_brush_cycler(0))
446 elif intra_value == 1:
447 sectors[zone].setBrush(et.utils.pg_brush_cycler(1))
448 elif data.used_config == _Mode.WAKE_UP_CONFIG:
449 assert data.wake_up_detections is not None
450 if data.wake_up_detections[zone] > 0:
451 sectors[zone].setBrush(pg.mkBrush("#b5afa0"))
452 else:
453 assert data.max_presence_zone is not None
454 if data.max_presence_zone == data.max_intra_zone:
455 sectors[data.max_presence_zone].setBrush(et.utils.pg_brush_cycler(1))
456 else:
457 sectors[data.max_presence_zone].setBrush(et.utils.pg_brush_cycler(0))
458
459
460if __name__ == "__main__":
461 main()
View this example on GitHub: acconeer/acconeer-python-exploration