examples/a121/algo/presence/detector.py

examples/a121/algo/presence/detector.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._core.entities.configs.sensor_config import SensorConfig
 17from acconeer.exptool.a121.algo.presence import Detector, DetectorConfig
 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    detector_config = DetectorConfig(
 27        start_m=1.0,
 28        end_m=3.0,
 29    )
 30
 31    detector = Detector(client=client, sensor_id=1, detector_config=detector_config)
 32    detector.start()
 33
 34    pg_updater = PGUpdater(
 35        detector_config,
 36        detector_config.to_sensor_config(),
 37        detector.estimated_frame_rate,
 38    )
 39    pg_process = et.PGProcess(pg_updater)
 40    pg_process.start()
 41
 42    interrupt_handler = et.utils.ExampleInterruptHandler()
 43
 44    while not interrupt_handler.got_signal:
 45        detector_result = detector.get_next()
 46        s = "Presence! " if detector_result.presence_detected else "No presence. "
 47        s += (
 48            f"Intra presence score {detector_result.intra_presence_score:.3f}, "
 49            f"inter presence score {detector_result.inter_presence_score:.3f}, "
 50            f"presence at {detector_result.presence_distance:.3f} m"
 51        )
 52        print(s)
 53        try:
 54            pg_process.put_data(detector_result)
 55        except et.PGProccessDiedException:
 56            break
 57
 58    detector.stop()
 59
 60    print("Disconnecting...")
 61    client.close()
 62
 63
 64class PGUpdater:
 65    def __init__(
 66        self,
 67        detector_config: DetectorConfig,
 68        sensor_config: SensorConfig,
 69        estimated_frame_rate: float,
 70    ):
 71        self.detector_config = detector_config
 72        self.distances = np.linspace(
 73            detector_config.start_m, detector_config.end_m, sensor_config.num_points
 74        )
 75
 76        self.history_length_s = 5
 77        self.history_length_n = int(round(self.history_length_s * estimated_frame_rate))
 78        self.intra_history = np.zeros(self.history_length_n)
 79        self.inter_history = np.zeros(self.history_length_n)
 80
 81        self.setup_is_done = False
 82
 83    def setup(self, win):
 84        win.setWindowTitle("Acconeer presence detection example")
 85
 86        self.intra_limit_lines = []
 87        self.inter_limit_lines = []
 88
 89        # Noise estimation plot
 90
 91        self.noise_plot = win.addPlot(
 92            row=0,
 93            col=0,
 94            title="Noise",
 95        )
 96        self.noise_plot.setMenuEnabled(False)
 97        self.noise_plot.setMouseEnabled(x=False, y=False)
 98        self.noise_plot.hideButtons()
 99        self.noise_plot.showGrid(x=True, y=True)
100        self.noise_plot.setLabel("bottom", "Distance (m)")
101        self.noise_plot.setLabel("left", "Amplitude")
102        self.noise_plot.setVisible(False)
103        self.noise_curve = self.noise_plot.plot(pen=et.utils.pg_pen_cycler())
104        self.noise_smooth_max = et.utils.SmoothMax(self.detector_config.frame_rate)
105
106        # Depthwise presence plot
107
108        self.move_plot = pg.PlotItem(title="Depthwise presence")
109        self.move_plot.setMenuEnabled(False)
110        self.move_plot.setMouseEnabled(x=False, y=False)
111        self.move_plot.hideButtons()
112        self.move_plot.showGrid(x=True, y=True)
113        self.move_plot.setLabel("bottom", "Distance (m)")
114        self.move_plot.setLabel("left", "Norm. ampl.")
115        self.move_plot.setXRange(self.distances[0], self.distances[-1])
116        self.intra_curve = self.move_plot.plot(pen=et.utils.pg_pen_cycler(1))
117        if not self.detector_config.intra_enable:
118            self.intra_curve.hide()
119
120        self.inter_curve = self.move_plot.plot(pen=et.utils.pg_pen_cycler(0))
121        if not self.detector_config.inter_enable:
122            self.inter_curve.hide()
123
124        self.move_smooth_max = et.utils.SmoothMax(
125            self.detector_config.frame_rate,
126            tau_decay=1.0,
127            tau_grow=0.25,
128        )
129
130        self.move_depth_line = pg.InfiniteLine(pen=pg.mkPen("k", width=1.5))
131        self.move_depth_line.hide()
132        self.move_plot.addItem(self.move_depth_line)
133
134        self.present_html_format = (
135            '<div style="text-align: center">'
136            '<span style="color: #FFFFFF;font-size:15pt;">'
137            "{}</span></div>"
138        )
139        not_present_html = (
140            '<div style="text-align: center">'
141            '<span style="color: #FFFFFF;font-size:15pt;">'
142            "{}</span></div>".format("No presence detected")
143        )
144        self.present_text_item = pg.TextItem(
145            fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
146            anchor=(0.5, 0),
147        )
148        self.not_present_text_item = pg.TextItem(
149            html=not_present_html,
150            fill=pg.mkColor(0x1F, 0x77, 0xB4, 180),
151            anchor=(0.5, 0),
152        )
153
154        self.move_plot.addItem(self.present_text_item)
155        self.move_plot.addItem(self.not_present_text_item)
156        self.present_text_item.hide()
157        self.not_present_text_item.hide()
158
159        # Intra presence history plot
160
161        self.intra_hist_plot = win.addPlot(
162            row=1,
163            col=0,
164            title="Intra presence history (fast motions)",
165        )
166        self.intra_hist_plot.setMenuEnabled(False)
167        self.intra_hist_plot.setMouseEnabled(x=False, y=False)
168        self.intra_hist_plot.hideButtons()
169        self.intra_hist_plot.showGrid(x=True, y=True)
170        self.intra_hist_plot.setLabel("bottom", "Time (s)")
171        self.intra_hist_plot.setLabel("left", "Score")
172        self.intra_hist_plot.setXRange(-self.history_length_s, 0)
173        self.intra_history_smooth_max = et.utils.SmoothMax(self.detector_config.frame_rate)
174        self.intra_hist_plot.setYRange(0, 10)
175        if not self.detector_config.intra_enable:
176            intra_color = et.utils.color_cycler(1)
177            intra_color = f"{intra_color}50"
178            intra_dashed_pen = pg.mkPen(intra_color, width=2.5, style=QtCore.Qt.DashLine)
179            intra_pen = pg.mkPen(intra_color, width=2)
180        else:
181            intra_dashed_pen = et.utils.pg_pen_cycler(1, width=2.5, style="--")
182            intra_pen = et.utils.pg_pen_cycler(1)
183
184        self.intra_hist_curve = self.intra_hist_plot.plot(pen=intra_pen)
185        limit_line = pg.InfiniteLine(angle=0, pen=intra_dashed_pen)
186        self.intra_hist_plot.addItem(limit_line)
187        self.intra_limit_lines.append(limit_line)
188
189        for line in self.intra_limit_lines:
190            line.setPos(self.detector_config.intra_detection_threshold)
191
192        # Inter presence history plot
193
194        self.inter_hist_plot = win.addPlot(
195            row=1,
196            col=1,
197            title="Inter presence history (slow motions)",
198        )
199        self.inter_hist_plot.setMenuEnabled(False)
200        self.inter_hist_plot.setMouseEnabled(x=False, y=False)
201        self.inter_hist_plot.hideButtons()
202        self.inter_hist_plot.showGrid(x=True, y=True)
203        self.inter_hist_plot.setLabel("bottom", "Time (s)")
204        self.inter_hist_plot.setLabel("left", "Score")
205        self.inter_hist_plot.setXRange(-self.history_length_s, 0)
206        self.inter_history_smooth_max = et.utils.SmoothMax(self.detector_config.frame_rate)
207        self.inter_hist_plot.setYRange(0, 10)
208        if not self.detector_config.inter_enable:
209            inter_color = et.utils.color_cycler(0)
210            inter_color = f"{inter_color}50"
211            inter_dashed_pen = pg.mkPen(inter_color, width=2.5, style=QtCore.Qt.DashLine)
212            inter_pen = pg.mkPen(inter_color, width=2)
213        else:
214            inter_pen = et.utils.pg_pen_cycler(0)
215            inter_dashed_pen = et.utils.pg_pen_cycler(0, width=2.5, style="--")
216
217        self.inter_hist_curve = self.inter_hist_plot.plot(pen=inter_pen)
218        limit_line = pg.InfiniteLine(angle=0, pen=inter_dashed_pen)
219        self.inter_hist_plot.addItem(limit_line)
220        self.inter_limit_lines.append(limit_line)
221
222        for line in self.inter_limit_lines:
223            line.setPos(self.detector_config.inter_detection_threshold)
224
225        sublayout = win.addLayout(row=2, col=0, colspan=2)
226        sublayout.layout.setColumnStretchFactor(0, 2)
227        sublayout.addItem(self.move_plot, row=0, col=0)
228
229        self.setup_is_done = True
230
231    def update(self, data):
232        noise = data.processor_extra_result.lp_noise
233        self.noise_curve.setData(self.distances, noise)
234        self.noise_plot.setYRange(0, self.noise_smooth_max.update(noise))
235
236        movement_x = data.presence_distance
237
238        self.inter_curve.setData(self.distances, data.inter_depthwise_scores)
239        self.intra_curve.setData(self.distances, data.intra_depthwise_scores)
240        m = self.move_smooth_max.update(
241            np.max(np.maximum(data.inter_depthwise_scores, data.intra_depthwise_scores))
242        )
243        m = max(
244            m,
245            2
246            * np.maximum(
247                self.detector_config.intra_detection_threshold,
248                self.detector_config.inter_detection_threshold,
249            ),
250        )
251        self.move_plot.setYRange(0, m)
252        self.move_depth_line.setPos(movement_x)
253        self.move_depth_line.setVisible(bool(data.presence_detected))
254
255        self.set_present_text_y_pos(m)
256
257        if data.presence_detected:
258            present_text = f"Presence detected at {movement_x * 100:.0f} cm"
259            present_html = self.present_html_format.format(present_text)
260            self.present_text_item.setHtml(present_html)
261
262            self.present_text_item.show()
263            self.not_present_text_item.hide()
264        else:
265            self.present_text_item.hide()
266            self.not_present_text_item.show()
267
268        # Intra presence
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.detector_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.detector_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