examples/a121/algo/smart_presence/processor.py#
1# Copyright (c) Acconeer AB, 2023-2026
2# All rights reserved
3
4from __future__ import annotations
5
6import numpy as np
7import numpy.typing as npt
8
9from PySide6 import QtCore
10
11import pyqtgraph as pg
12
13import acconeer.exptool as et
14from acconeer.exptool import a121
15from acconeer.exptool.a121.algo import presence, smart_presence
16
17
18def main():
19 args = a121.ExampleArgumentParser().parse_args()
20 et.utils.config_logging(args)
21
22 client = a121.Client.open(**a121.get_client_args(args))
23
24 processor_config = smart_presence.ProcessorConfig(
25 num_zones=3,
26 )
27
28 detector_config = presence.DetectorConfig(
29 start_m=1,
30 end_m=3,
31 )
32
33 presence_detector = presence.Detector(
34 client=client, sensor_id=1, detector_config=detector_config
35 )
36 presence_detector.start()
37
38 smart_presence_processor = smart_presence.Processor(
39 processor_config,
40 detector_config,
41 presence_detector.session_config,
42 presence_detector.detector_metadata,
43 )
44
45 pg_updater = PGUpdater(
46 detector_config,
47 processor_config,
48 detector_config.to_sensor_config(),
49 presence_detector.estimated_frame_rate,
50 smart_presence_processor.zone_limits,
51 )
52 pg_process = et.PGProcess(pg_updater)
53 pg_process.start()
54
55 interrupt_handler = et.utils.ExampleInterruptHandler()
56
57 while not interrupt_handler.got_signal:
58 detector_result = presence_detector.get_next()
59 processor_result = smart_presence_processor.process(detector_result)
60 if detector_result.presence_detected:
61 print(f"Presence in zone {processor_result.max_presence_zone}")
62 else:
63 print("No presence")
64
65 data = {
66 "processor_result": processor_result,
67 "detector_result": detector_result,
68 }
69 try:
70 pg_process.put_data(data)
71 except et.PGProccessDiedException:
72 break
73
74 presence_detector.stop()
75
76 print("Disconnecting...")
77 client.close()
78
79
80class PGUpdater:
81 def __init__(
82 self,
83 detector_config: presence.DetectorConfig,
84 processor_config: smart_presence.ProcessorConfig,
85 sensor_config: a121.SensorConfig,
86 estimated_frame_rate: float,
87 zone_limits: npt.NDArray[np.float64],
88 ):
89 self.detector_config = detector_config
90
91 self.history_length_s = 5
92 self.estimated_frame_rate = estimated_frame_rate
93 self.history_length_n = int(round(self.history_length_s * estimated_frame_rate))
94 self.intra_history = np.zeros(self.history_length_n)
95 self.inter_history = np.zeros(self.history_length_n)
96
97 self.num_sectors = min(processor_config.num_zones, sensor_config.num_points)
98 self.sector_size = max(1, -(-sensor_config.num_points // self.num_sectors))
99
100 self.sector_offset = (self.num_sectors * self.sector_size - sensor_config.num_points) // 2
101 self.zone_limits = zone_limits
102
103 self.setup_is_done = False
104
105 def setup(self, win):
106 win.setWindowTitle("Acconeer smart presence example")
107
108 self.intra_limit_lines = []
109 self.inter_limit_lines = []
110
111 # Intra presence history plot
112
113 self.intra_hist_plot = win.addPlot(
114 row=0,
115 col=0,
116 title="Intra presence history (fast motions)",
117 )
118 self.intra_hist_plot.setMenuEnabled(False)
119 self.intra_hist_plot.setMouseEnabled(x=False, y=False)
120 self.intra_hist_plot.hideButtons()
121 self.intra_hist_plot.showGrid(x=True, y=True)
122 self.intra_hist_plot.setLabel("bottom", "Time (s)")
123 self.intra_hist_plot.setLabel("left", "Score")
124 self.intra_hist_plot.setXRange(-self.history_length_s, 0)
125 self.intra_history_smooth_max = et.utils.SmoothMax(self.estimated_frame_rate)
126 self.intra_hist_plot.setYRange(0, 10)
127 if not self.detector_config.intra_enable:
128 intra_color = et.utils.color_cycler(1)
129 intra_color = f"{intra_color}50"
130 intra_dashed_pen = pg.mkPen(intra_color, width=2.5, style=QtCore.Qt.DashLine)
131 intra_pen = pg.mkPen(intra_color, width=2)
132 else:
133 intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
134 intra_pen = et.utils.pg_pen_cycler(1)
135
136 self.intra_hist_curve = self.intra_hist_plot.plot(pen=intra_pen)
137 limit_line = pg.InfiniteLine(angle=0, pen=intra_dashed_pen)
138 self.intra_hist_plot.addItem(limit_line)
139 self.intra_limit_lines.append(limit_line)
140
141 for line in self.intra_limit_lines:
142 line.setPos(self.detector_config.intra_detection_threshold)
143
144 # Inter presence history plot
145
146 self.inter_hist_plot = win.addPlot(
147 row=0,
148 col=1,
149 title="Inter presence history (slow motions)",
150 )
151 self.inter_hist_plot.setMenuEnabled(False)
152 self.inter_hist_plot.setMouseEnabled(x=False, y=False)
153 self.inter_hist_plot.hideButtons()
154 self.inter_hist_plot.showGrid(x=True, y=True)
155 self.inter_hist_plot.setLabel("bottom", "Time (s)")
156 self.inter_hist_plot.setLabel("left", "Score")
157 self.inter_hist_plot.setXRange(-self.history_length_s, 0)
158 self.inter_history_smooth_max = et.utils.SmoothMax(self.estimated_frame_rate)
159 self.inter_hist_plot.setYRange(0, 10)
160 if not self.detector_config.inter_enable:
161 inter_color = et.utils.color_cycler(0)
162 inter_color = f"{inter_color}50"
163 inter_dashed_pen = pg.mkPen(inter_color, width=2.5, style=QtCore.Qt.DashLine)
164 inter_pen = pg.mkPen(inter_color, width=2)
165 else:
166 inter_pen = et.utils.pg_pen_cycler(0)
167 inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
168
169 self.inter_hist_curve = self.inter_hist_plot.plot(pen=inter_pen)
170 limit_line = pg.InfiniteLine(angle=0, pen=inter_dashed_pen)
171 self.inter_hist_plot.addItem(limit_line)
172 self.inter_limit_lines.append(limit_line)
173
174 for line in self.inter_limit_lines:
175 line.setPos(self.detector_config.inter_detection_threshold)
176
177 # Sector plot
178
179 self.sector_plot = pg.PlotItem(
180 title="Detection zone<br>Detection type: fast (orange), slow (blue), both (green)"
181 )
182 self.sector_plot.setAspectLocked()
183 self.sector_plot.hideAxis("left")
184 self.sector_plot.hideAxis("bottom")
185 self.sectors = []
186 self.limit_text = []
187
188 self.range_html = (
189 '<div style="text-align: center">'
190 '<span style="color: #000000;font-size:12pt;">'
191 "{}</span></div>"
192 )
193
194 pen = pg.mkPen("k", width=1)
195 span_deg = 25
196 for r in np.flip(np.arange(self.num_sectors) + 1):
197 sector = pg.QtWidgets.QGraphicsEllipseItem(-r, -r, r * 2, r * 2)
198 sector.setStartAngle(-16 * span_deg)
199 sector.setSpanAngle(16 * span_deg * 2)
200 sector.setPen(pen)
201 self.sector_plot.addItem(sector)
202 self.sectors.append(sector)
203
204 limit = pg.TextItem(html=self.range_html, anchor=(0.5, 0.5), angle=25)
205 x = r * np.cos(np.radians(span_deg))
206 y = r * np.sin(np.radians(span_deg))
207 limit.setPos(x, y + 0.25)
208 self.sector_plot.addItem(limit)
209 self.limit_text.append(limit)
210
211 self.sectors.reverse()
212
213 start_limit_text = pg.TextItem(html=self.range_html, anchor=(0.5, 0.5), angle=25)
214 range_html = self.range_html.format(f"{self.detector_config.start_m}")
215 start_limit_text.setHtml(range_html)
216 start_limit_text.setPos(0, 0.25)
217 self.sector_plot.addItem(start_limit_text)
218
219 unit_text = pg.TextItem(html=self.range_html, anchor=(0.5, 0.5))
220 unit_html = self.range_html.format("[m]")
221 unit_text.setHtml(unit_html)
222 unit_text.setPos(
223 self.num_sectors + 0.5, (self.num_sectors + 1) * np.sin(np.radians(span_deg))
224 )
225 self.sector_plot.addItem(unit_text)
226
227 for text_item, limit in zip(self.limit_text, np.flip(self.zone_limits)):
228 range_html = self.range_html.format(np.around(limit, 1))
229 text_item.setHtml(range_html)
230
231 sublayout = win.addLayout(row=1, col=0, colspan=2)
232 sublayout.layout.setColumnStretchFactor(0, 2)
233 sublayout.addItem(self.sector_plot, row=0, col=0)
234
235 self.setup_is_done = True
236
237 def update(self, data):
238 processor_result = data["processor_result"]
239 detector_result = data["detector_result"]
240
241 # Intra presence
242
243 move_hist_xs = np.linspace(-self.history_length_s, 0, self.history_length_n)
244
245 self.intra_history = np.roll(self.intra_history, -1)
246 self.intra_history[-1] = detector_result.intra_presence_score
247
248 m_hist = max(
249 float(np.max(self.intra_history)),
250 self.detector_config.intra_detection_threshold * 1.05,
251 )
252 m_hist = self.intra_history_smooth_max.update(m_hist)
253
254 self.intra_hist_plot.setYRange(0, m_hist)
255 self.intra_hist_curve.setData(move_hist_xs, self.intra_history)
256
257 # Inter presence
258
259 self.inter_history = np.roll(self.inter_history, -1)
260 self.inter_history[-1] = detector_result.inter_presence_score
261
262 m_hist = max(
263 float(np.max(self.inter_history)),
264 self.detector_config.inter_detection_threshold * 1.05,
265 )
266 m_hist = self.inter_history_smooth_max.update(m_hist)
267
268 self.inter_hist_plot.setYRange(0, m_hist)
269 self.inter_hist_curve.setData(move_hist_xs, self.inter_history)
270
271 # Sector
272
273 brush = et.utils.pg_brush_cycler(7)
274 for sector in self.sectors:
275 sector.setBrush(brush)
276
277 if detector_result.presence_detected:
278 assert processor_result.max_presence_zone is not None
279 if processor_result.max_presence_zone == processor_result.max_intra_zone:
280 self.sectors[processor_result.max_presence_zone].setBrush(
281 et.utils.pg_brush_cycler(1)
282 )
283 else:
284 self.sectors[processor_result.max_presence_zone].setBrush(
285 et.utils.pg_brush_cycler(0)
286 )
287
288
289if __name__ == "__main__":
290 main()
View this example on GitHub: acconeer/acconeer-python-exploration