examples/a121/algo/breathing/breathing_with_gui.py

examples/a121/algo/breathing/breathing_with_gui.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.QtGui import QFont
 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.breathing import AppState, RefApp
 18from acconeer.exptool.a121.algo.breathing._ref_app import (
 19    BreathingProcessorConfig,
 20    RefAppConfig,
 21    get_sensor_config,
 22)
 23from acconeer.exptool.a121.algo.presence import ProcessorConfig as PresenceProcessorConfig
 24
 25
 26def main():
 27    args = a121.ExampleArgumentParser().parse_args()
 28    et.utils.config_logging(args)
 29
 30    # Setup the configurations
 31    # Detailed at https://docs.acconeer.com/en/latest/exploration_tool/algo/a121/ref_apps/breathing.html
 32
 33    # Sensor selections
 34    sensor = 1
 35
 36    # Ref App Configurations
 37    breathing_processor_config = BreathingProcessorConfig(
 38        lowest_breathing_rate=6,
 39        highest_breathing_rate=60,
 40        time_series_length_s=20,
 41    )
 42
 43    # Presence Configurations
 44    presence_config = PresenceProcessorConfig(
 45        intra_detection_threshold=4,
 46        intra_frame_time_const=0.15,
 47        inter_frame_fast_cutoff=20,
 48        inter_frame_slow_cutoff=0.2,
 49        inter_frame_deviation_time_const=0.5,
 50    )
 51
 52    # Breathing Configurations
 53    ref_app_config = RefAppConfig(
 54        use_presence_processor=True,
 55        num_distances_to_analyze=3,
 56        distance_determination_duration=5,
 57        breathing_config=breathing_processor_config,
 58        presence_config=presence_config,
 59    )
 60
 61    # End setup configurations
 62
 63    # Preparation for client
 64    sensor_config = get_sensor_config(ref_app_config=ref_app_config)
 65    client = a121.Client.open(**a121.get_client_args(args))
 66    metadata = client.setup_session(sensor_config)
 67
 68    # Preparation for reference application processor
 69    ref_app = RefApp(client=client, sensor_id=sensor, ref_app_config=ref_app_config)
 70    ref_app.start()
 71
 72    pg_updater = PGUpdater(sensor_config, ref_app_config, metadata)
 73    pg_process = et.PGProcess(pg_updater)
 74    pg_process.start()
 75
 76    interrupt_handler = et.utils.ExampleInterruptHandler()
 77
 78    while not interrupt_handler.got_signal:
 79        processed_data = ref_app.get_next()
 80        try:
 81            pg_process.put_data(processed_data)
 82        except et.PGProccessDiedException:
 83            break
 84
 85    ref_app.stop()
 86    print("Disconnecting...")
 87    client.close()
 88
 89
 90class PGUpdater:
 91    def __init__(
 92        self,
 93        sensor_config: a121.SensorConfig,
 94        ref_app_config: RefAppConfig,
 95        metadata: a121.Metadata,
 96    ):
 97        self.distances = get_distances_m(sensor_config, metadata)
 98        self.use_presence_processor = ref_app_config.use_presence_processor
 99
100    def setup(self, win):
101        # Define pens and font.
102        blue_color = et.utils.color_cycler(0)
103        orange_color = et.utils.color_cycler(1)
104        brush = et.utils.pg_brush_cycler(0)
105        self.blue = dict(
106            pen=pg.mkPen(blue_color, width=2),
107            symbol="o",
108            symbolSize=1,
109            symbolBrush=brush,
110            symbolPen="k",
111        )
112        self.orange = dict(
113            pen=pg.mkPen(orange_color, width=2),
114            symbol="o",
115            symbolSize=1,
116            symbolBrush=brush,
117            symbolPen="k",
118        )
119        self.blue_transparent_pen = pg.mkPen(f"{blue_color}50", width=2)
120        self.orange_transparent_pen = pg.mkPen(f"{orange_color}50", width=2)
121
122        brush_dot = et.utils.pg_brush_cycler(1)
123        symbol_dot_kw = dict(symbol="o", symbolSize=10, symbolBrush=brush_dot, symbolPen="k")
124
125        font = QFont()
126        font.setPixelSize(16)
127
128        # Presence plot.
129        self.presence_plot = win.addPlot(row=0, col=0)
130        self.presence_plot.setMenuEnabled(False)
131        self.presence_plot.showGrid(x=True, y=True)
132        self.presence_plot.addLegend()
133        self.presence_plot.setLabel("left", "Presence score")
134        self.presence_plot.setLabel("bottom", "Distance (m)")
135        self.presence_plot.addItem(pg.PlotDataItem())
136        self.presence_plot_curve = []
137        self.presence_plot_curve.append(self.presence_plot.plot(**self.blue))
138        self.presence_plot_curve.append(self.presence_plot.plot(**self.orange))
139        self.presence_plot_curve.append(self.presence_plot.plot(**self.blue))
140        self.presence_plot_curve.append(self.presence_plot.plot(**self.orange))
141
142        self.presence_plot_legend = pg.LegendItem(offset=(0.0, 0.5))
143        self.presence_plot_legend.setParentItem(self.presence_plot)
144        self.presence_plot_legend.addItem(self.presence_plot_curve[2], "Slow motion")
145        self.presence_plot_legend.addItem(self.presence_plot_curve[3], "Fast motion")
146        self.presence_plot_legend.show()
147
148        self.presence_smoot_max = et.utils.SmoothMax()
149
150        self.presence_text_item = pg.TextItem(
151            fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
152            anchor=(0.5, 0),
153            color=pg.mkColor(0xFF, 0xFF, 0xFF, 200),
154        )
155        self.presence_text_item.setFont(font)
156        self.presence_text_item.show()
157        self.presence_plot.addItem(self.presence_text_item)
158
159        # Time series plot.
160        self.time_series_plot = win.addPlot(row=1, col=0)
161        self.time_series_plot.setMenuEnabled(False)
162        self.time_series_plot.showGrid(x=True, y=True)
163        self.time_series_plot.addLegend()
164        self.time_series_plot.setLabel("left", "Displacement")
165        self.time_series_plot.setLabel("bottom", "Time (s)")
166        self.time_series_plot.addItem(pg.PlotDataItem())
167        self.time_series_curve = self.time_series_plot.plot(**self.blue)
168
169        self.time_series_text_item = pg.TextItem(
170            fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
171            anchor=(0.5, 0),
172            color=pg.mkColor(0xFF, 0xFF, 0xFF, 200),
173        )
174        self.time_series_text_item.setFont(font)
175        self.time_series_text_item.show()
176        self.time_series_plot.addItem(self.time_series_text_item)
177
178        # Breathing psd plot.
179        self.breathing_psd_plot = win.addPlot(row=2, col=0)
180        self.breathing_psd_plot.setMenuEnabled(False)
181        self.breathing_psd_plot.showGrid(x=True, y=True)
182        self.breathing_psd_plot.addLegend()
183        self.breathing_psd_plot.setLabel("left", "PSD")
184        self.breathing_psd_plot.setLabel("bottom", "Breathing rate (Hz)")
185        self.breathing_psd_plot.addItem(pg.PlotDataItem())
186        self.breathing_psd_curve = self.breathing_psd_plot.plot(**self.blue)
187
188        self.psd_smoothing = et.utils.SmoothMax()
189
190        # Breathing rate plot.
191        self.breathing_rate_plot = win.addPlot(row=3, col=0)
192        self.breathing_rate_plot.setMenuEnabled(False)
193        self.breathing_rate_plot.showGrid(x=True, y=True)
194        self.breathing_rate_plot.addLegend()
195        self.breathing_rate_plot.setLabel("left", "Breaths per minute")
196        self.breathing_rate_plot.setLabel("bottom", "Time (s)")
197        self.breathing_rate_plot.addItem(pg.PlotDataItem())
198        self.breathing_rate_curves = []
199        self.breathing_rate_curves.append(self.breathing_rate_plot.plot(**self.blue))
200        self.breathing_rate_curves.append(
201            self.breathing_rate_plot.plot(**dict(pen=None, **symbol_dot_kw))
202        )
203        self.smooth_breathing_rate = et.utils.SmoothLimits()
204
205        self.breathing_rate_plot_legend = pg.LegendItem(offset=(0.0, 0.5))
206        self.breathing_rate_plot_legend.setParentItem(self.breathing_rate_plot)
207        self.breathing_rate_plot_legend.addItem(self.breathing_rate_curves[0], "Breathing rate")
208        self.breathing_rate_plot_legend.addItem(
209            self.breathing_rate_curves[1], "Breathing rate(embedded output)"
210        )
211        self.breathing_rate_plot_legend.hide()
212
213        self.breathing_rate_text_item = pg.TextItem(
214            fill=pg.mkColor(0xFF, 0x7F, 0x0E, 200),
215            anchor=(0.5, 0),
216            color=pg.mkColor(0xFF, 0xFF, 0xFF, 200),
217        )
218        self.breathing_rate_text_item.setFont(font)
219        self.breathing_rate_text_item.hide()
220        self.breathing_rate_plot.addItem(self.breathing_rate_text_item)
221
222    def update(self, ref_app_result):
223        app_state = ref_app_result.app_state
224
225        max_ampl = max(
226            np.max(ref_app_result.presence_result.inter),
227            np.max(ref_app_result.presence_result.intra),
228        )
229        lim = self.presence_smoot_max.update(max_ampl)
230        self.presence_plot.setYRange(0, lim)
231
232        if ref_app_result.distances_being_analyzed is None:
233            self.presence_plot_curve[0].setData(
234                self.distances, ref_app_result.presence_result.inter, **self.blue
235            )
236            self.presence_plot_curve[1].setData(
237                self.distances, ref_app_result.presence_result.intra, **self.orange
238            )
239            self.presence_plot_curve[2].setData([], [])
240            self.presence_plot_curve[3].setData([], [])
241        else:
242            start = ref_app_result.distances_being_analyzed[0]
243            end = ref_app_result.distances_being_analyzed[1]
244            s = slice(start, end)
245            distance_slice = self.distances[s]
246            self.presence_plot_curve[0].setData(
247                self.distances,
248                ref_app_result.presence_result.inter,
249                pen=self.blue_transparent_pen,
250            )
251            self.presence_plot_curve[1].setData(
252                self.distances,
253                ref_app_result.presence_result.intra,
254                pen=self.orange_transparent_pen,
255            )
256            self.presence_plot_curve[2].setData(
257                distance_slice, ref_app_result.presence_result.inter[s]
258            )
259            self.presence_plot_curve[3].setData(
260                distance_slice, ref_app_result.presence_result.intra[s]
261            )
262
263        if ref_app_result.breathing_result is not None:
264            breathing_result = ref_app_result.breathing_result.extra_result
265            breathing_motion = breathing_result.breathing_motion
266            psd = breathing_result.psd
267            frequencies = breathing_result.frequencies
268            time_vector = breathing_result.time_vector
269            all_breathing_rate_history = breathing_result.all_breathing_rate_history
270            breathing_rate_history = breathing_result.breathing_rate_history
271
272            self.time_series_curve.setData(
273                time_vector[-breathing_motion.shape[0] :], breathing_motion
274            )
275            y = np.max(np.abs(breathing_motion)) * 1.05
276            self.time_series_plot.setYRange(-y, y)
277            self.time_series_plot.setXRange(
278                time_vector[-breathing_motion.shape[0]], max(time_vector)
279            )
280
281            if not np.all(np.isnan(all_breathing_rate_history)):
282                ylim = self.psd_smoothing.update(psd)
283                self.breathing_psd_curve.setData(frequencies, psd)
284                self.breathing_psd_plot.setYRange(0, ylim)
285                self.breathing_psd_plot.setXRange(0, 2)
286
287                self.breathing_rate_curves[0].setData(time_vector, all_breathing_rate_history)
288                lims = self.smooth_breathing_rate.update(all_breathing_rate_history)
289                self.breathing_rate_plot.setYRange(lims[0] - 3, lims[1] + 3)
290
291                self.breathing_rate_plot_legend.show()
292
293            if not np.all(np.isnan(breathing_rate_history)):
294                self.breathing_rate_curves[1].setData(time_vector, breathing_rate_history)
295
296            if not np.isnan(breathing_rate_history[-1]):
297                self.displayed_breathing_rate = f"{breathing_rate_history[-1]:.1f}"
298                self.breathing_rate_text_item.show()
299
300        else:
301            self.time_series_plot.setYRange(0, 1)
302            self.time_series_plot.setXRange(0, 1)
303            self.time_series_curve.setData([], [])
304            self.breathing_psd_curve.setData([], [])
305            self.breathing_rate_curves[0].setData([], [])
306            self.breathing_rate_curves[1].setData([], [])
307            self.displayed_breathing_rate = None
308            self.breathing_rate_text_item.hide()
309
310        # Set text in text boxes according to app state.
311
312        # Presence text
313        if app_state == AppState.NO_PRESENCE_DETECTED:
314            presence_text = "No presence detected"
315        elif app_state == AppState.DETERMINE_DISTANCE_ESTIMATE:
316            presence_text = "Determining distance with presence"
317        elif app_state == AppState.ESTIMATE_BREATHING_RATE:
318            start_m = f"{distance_slice[0]:.2f}"
319            end_m = f"{distance_slice[-1]:.2f}"
320            if self.use_presence_processor:
321                presence_text = (
322                    "Presence detected in the range " + start_m + " - " + end_m + " (m)"
323                )
324            else:
325                presence_text = "Presence distance detection disabled"
326        elif app_state == AppState.INTRA_PRESENCE_DETECTED:
327            presence_text = "Large motion detected"
328        else:
329            presence_text = ""
330
331        text_y_pos = self.presence_plot.getAxis("left").range[1] * 0.95
332        text_x_pos = (
333            self.presence_plot.getAxis("bottom").range[1]
334            + self.presence_plot.getAxis("bottom").range[0]
335        ) / 2.0
336        self.presence_text_item.setPos(text_x_pos, text_y_pos)
337        self.presence_text_item.setHtml(presence_text)
338
339        # Breathing text
340        if app_state == AppState.ESTIMATE_BREATHING_RATE:
341            if (
342                ref_app_result.breathing_result is not None
343                and ref_app_result.breathing_result.breathing_rate is None
344            ):
345                time_series_text = "Initializing breathing detection"
346            elif self.displayed_breathing_rate is not None:
347                time_series_text = "Breathing rate: " + self.displayed_breathing_rate + " bpm"
348            else:
349                time_series_text = ""
350        else:
351            time_series_text = "Waiting for distance"
352
353        text_y_pos = self.time_series_plot.getAxis("left").range[1] * 0.95
354        text_x_pos = (
355            self.time_series_plot.getAxis("bottom").range[1]
356            + self.time_series_plot.getAxis("bottom").range[0]
357        ) / 2.0
358        self.time_series_text_item.setPos(text_x_pos, text_y_pos)
359        self.time_series_text_item.setHtml(time_series_text)
360
361        if self.displayed_breathing_rate is not None:
362            text_y_pos = self.breathing_rate_plot.getAxis("left").range[1] * 0.95
363            text_x_pos = time_vector[0]
364
365            self.breathing_rate_text_item.setPos(text_x_pos, text_y_pos)
366            self.breathing_rate_text_item.setHtml(self.displayed_breathing_rate + " bpm")
367
368
369if __name__ == "__main__":
370    main()

View this example on GitHub: acconeer/acconeer-python-exploration